gh RecordConsumer - add visibility to extra properties setup with aut…

…o.offset.reset (#35848) GitOrigin-RevId: 7b878399acd1e25003ccae071e817ed22b2fba21
wix · Oct 5, 2023 · 7c4bb05 · 7c4bb05
1 parent 97179da
commit 7c4bb05
Showing 1 changed file with 11 additions and 4 deletions.
diff --git a/core/src/main/scala/com/wixpress/dst/greyhound/core/consumer/RecordConsumer.scala b/core/src/main/scala/com/wixpress/dst/greyhound/core/consumer/RecordConsumer.scala
@@ -62,16 +62,23 @@ object RecordConsumer {
    * concurrent between partitions; order is guaranteed to be maintained within the same partition.
    */
   def make[R, E](
-                  config: RecordConsumerConfig,
-                  handler: RecordHandler[R, E, Chunk[Byte], Chunk[Byte]],
-                  createConsumerOverride: Option[ConsumerConfig => RIO[GreyhoundMetrics with Scope, Consumer]] = None
+    config: RecordConsumerConfig,
+    handler: RecordHandler[R, E, Chunk[Byte], Chunk[Byte]],
+    createConsumerOverride: Option[ConsumerConfig => RIO[GreyhoundMetrics with Scope, Consumer]] = None
   )(implicit trace: Trace, tag: Tag[Env]): ZIO[R with Env with Scope with GreyhoundMetrics, Throwable, RecordConsumer[R with Env]] =
     ZIO
       .acquireRelease(
         for {
           consumerShutdown <- AwaitShutdown.make
           _                <- GreyhoundMetrics
-                                .report(CreatingConsumer(config.clientId, config.group, config.bootstrapServers, config.consumerAttributes))
+                                .report(
+                                  CreatingConsumer(
+                                    config.clientId,
+                                    config.group,
+                                    config.bootstrapServers,
+                                    config.consumerAttributes ++ config.extraProperties
+                                  )
+                                )
 
           _                                    <- validateRetryPolicy(config)
           consumerSubscriptionRef              <- Ref.make[ConsumerSubscription](config.initialSubscription)