Merge pull request #476 from Banno/qwer

feat: configs seam for `RecordStream.Batched.assign` (-> `master`)
Banno · Aug 19, 2021 · 3619ad4 · 3619ad4
2 parents 37349c3 + b1b6e1a
commit 3619ad4
Showing 1 changed file with 49 additions and 0 deletions.
diff --git a/core/src/main/scala/com/banno/kafka/RecordStream.scala b/core/src/main/scala/com/banno/kafka/RecordStream.scala
@@ -669,5 +669,54 @@ object RecordStream {
           WhetherCommits.May(groupId),
         )
       )
+
+    def assign[F[_]: Async](
+        kafkaBootstrapServers: BootstrapServers,
+        schemaRegistryUri: SchemaRegistryUrl,
+        clientId: String,
+        configs: Map[String, AnyRef],
+    ): Assigner[F, IncomingRecords, Stream] =
+      AssignerImpl(
+        BaseConfigs(
+          kafkaBootstrapServers,
+          schemaRegistryUri,
+          clientId,
+          WhetherCommits.No,
+          configs.toList: _*
+        )
+      )
+
+    def assign[F[_]: Async](
+        kafkaBootstrapServers: BootstrapServers,
+        schemaRegistryUri: SchemaRegistryUrl,
+        groupId: GroupId,
+        configs: Map[String, AnyRef],
+    ): Assigner[F, IncomingRecords, RecordStream] =
+      AssignerImpl(
+        BaseConfigs(
+          kafkaBootstrapServers,
+          schemaRegistryUri,
+          groupId.id,
+          WhetherCommits.May(groupId),
+          configs.toList: _*
+        )
+      )
+
+    def assign[F[_]: Async](
+        kafkaBootstrapServers: BootstrapServers,
+        schemaRegistryUri: SchemaRegistryUrl,
+        clientId: String,
+        groupId: GroupId,
+        configs: Map[String, AnyRef],
+    ): Assigner[F, IncomingRecords, RecordStream] =
+      AssignerImpl(
+        BaseConfigs(
+          kafkaBootstrapServers,
+          schemaRegistryUri,
+          clientId,
+          WhetherCommits.May(groupId),
+          configs.toList: _*
+        )
+      )
   }
 }