From ab2b88e1d4ae40cafcfcb41c733ec71a0eb7da90 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alejandro=20Herna=CC=81ndez?= Date: Tue, 29 Sep 2026 11:45:52 +0200 Subject: [PATCH] Restore the consumer.group span attribute fs2-kafka 4 removed `CommittableOffset#consumerGroupId`, so the migration in #293 dropped the `consumer.group` attribute from `kafka.receive` spans. The group id is still reachable through `KafkaCommitter#metadata`, which calls the blocking `groupMetadata()` on the underlying consumer, so it is fetched once per committer and cached for the rest of the stream. A failing fetch (no `group.id` configured) falls back to an empty string, as before. --- .../trace4cats/kafka/TracedConsumer.scala | 18 +++++- .../trace4cats/kafka/TracedConsumerSpec.scala | 60 ++++++++++++++++++- 2 files changed, 75 insertions(+), 3 deletions(-) diff --git a/modules/trace4cats-kafka-client/src/main/scala/trace4cats/kafka/TracedConsumer.scala b/modules/trace4cats-kafka-client/src/main/scala/trace4cats/kafka/TracedConsumer.scala index c07b04b..4373ee8 100644 --- a/modules/trace4cats-kafka-client/src/main/scala/trace4cats/kafka/TracedConsumer.scala +++ b/modules/trace4cats-kafka-client/src/main/scala/trace4cats/kafka/TracedConsumer.scala @@ -24,6 +24,7 @@ package trace4cats.kafka import cats.Functor import cats.data.WriterT import cats.effect.kernel.MonadCancelThrow +import cats.syntax.applicativeError._ import cats.syntax.functor._ import fs2.Stream @@ -47,13 +48,26 @@ object TracedConsumer extends Fs2StreamSyntax { k: ResourceKleisli[F, SpanParams, Span[F]] )(implicit P: Provide[F, G, Span[F]]): TracedStream[F, CommittableConsumerRecord[F, K, V]] = stream - .traceContinue(k, "kafka.receive", SpanKind.Consumer) { record => + .evalMapAccumulate(Map.empty[KafkaCommitter[F], String]) { case (groups, record) => + val committer = record.offset.committer + + groups.get(committer) match { + case Some(group) => MonadCancelThrow[F].pure((groups, (record, group))) + case None => + committer.metadata.map(_.groupId).handleError(_ => "").map { group => + (groups.updated(committer, group), (record, group)) + } + } + } + .map(_._2) + .traceContinue(k, "kafka.receive", SpanKind.Consumer) { case (record, _) => KafkaHeaders.converter.from(record.record.headers) } - .evalMapTrace { record => + .evalMapTrace { case (record, group) => Trace[G] .putAll( "topic" -> record.record.topic, + "consumer.group" -> AttributeValue.StringValue(group), "create.time" -> AttributeValue.LongValue(createTime(record.record.timestamp)), "log.append.time" -> AttributeValue.LongValue(logAppendTime(record.record.timestamp)) ) diff --git a/modules/trace4cats-kafka-client/src/test/scala/trace4cats/kafka/TracedConsumerSpec.scala b/modules/trace4cats-kafka-client/src/test/scala/trace4cats/kafka/TracedConsumerSpec.scala index 6731b8a..839f032 100644 --- a/modules/trace4cats-kafka-client/src/test/scala/trace4cats/kafka/TracedConsumerSpec.scala +++ b/modules/trace4cats-kafka-client/src/test/scala/trace4cats/kafka/TracedConsumerSpec.scala @@ -21,9 +21,13 @@ package trace4cats.kafka +import scala.annotation.nowarn + import cats.data.Kleisli +import cats.data.NonEmptyList import cats.effect.IO import cats.effect.Ref +import cats.effect.Resource import cats.effect.unsafe.implicits.global import fs2.Stream @@ -31,16 +35,70 @@ import fs2.kafka.CommittableConsumerRecord import fs2.kafka.CommittableOffsetBatch import fs2.kafka.ConsumerRecord import fs2.kafka.TestCommittables +import org.apache.kafka.clients.consumer.ConsumerGroupMetadata import org.apache.kafka.clients.consumer.OffsetAndMetadata import org.apache.kafka.common.TopicPartition import org.scalatest.flatspec.AnyFlatSpec import trace4cats.EntryPoint import trace4cats.Span +import trace4cats.kernel.ErrorHandler +import trace4cats.model.AttributeValue +import trace4cats.model.Link +import trace4cats.model.SpanKind +import trace4cats.model.SpanStatus class TracedConsumerSpec extends AnyFlatSpec { type Traced[A] = Kleisli[IO, Span[IO], A] + @nowarn("cat=deprecation") + val metadata = new ConsumerGroupMetadata("my-group") + + behavior.of("TracedConsumer.inject") + + it should "set consumer.group from the committer metadata, fetched once per committer" in { + val partition = new TopicPartition("topic", 0) + + val test = for { + groups <- Ref.of[IO, List[Option[Any]]](Nil) + fetches <- Ref.of[IO, Int](0) + span <- IO.pure[Span[IO]](new Span[IO] { + private val noop = Span.noopInstance[IO] + + override def context = noop.context + override def put(key: String, value: AttributeValue) = noop.put(key, value) + override def putAll(fields: Map[String, AttributeValue]) = noop.putAll(fields) + override def setStatus(spanStatus: SpanStatus) = noop.setStatus(spanStatus) + override def addLink(link: Link) = noop.addLink(link) + override def addLinks(links: NonEmptyList[Link]) = noop.addLinks(links) + override def child(name: String, kind: SpanKind) = noop.child(name, kind) + override def child(name: String, kind: SpanKind, errorHandler: ErrorHandler) = + noop.child(name, kind, errorHandler) + + override def putAll(fields: (String, AttributeValue)*) = + groups.update(fields.toMap.get("consumer.group").map(_.value.value) :: _) + }) + committer = TestCommittables.committer[IO](_ => IO.unit, fetches.update(_ + 1).as(metadata)) + records = List(1L, 2L, 3L).map { offset => + CommittableConsumerRecord( + ConsumerRecord(partition.topic, partition.partition, offset, "key", "value"), + TestCommittables.offset(partition, offset, committer) + ) + } + _ <- TracedConsumer + .inject[IO, Traced, String, String](Stream.emits(records))(Kleisli(_ => Resource.pure[IO, Span[IO]](span))) + .run + .compile + .drain + recorded <- groups.get + count <- fetches.get + } yield assertResult((List.fill(3)(Some("my-group")), 1)) { + (recorded, count) + } + + test.unsafeRunSync() + } + behavior.of("TracedConsumer.injectK") it should "keep one lifted committer per consumer so batched commits stay merged" in { @@ -49,7 +107,7 @@ class TracedConsumerSpec extends AnyFlatSpec { val test = for { commits <- Ref.of[IO, List[Map[TopicPartition, OffsetAndMetadata]]](Nil) - committer = TestCommittables.committer[IO](offsets => commits.update(offsets :: _), IO.never) + committer = TestCommittables.committer[IO](offsets => commits.update(offsets :: _), IO.pure(metadata)) records = List((partition0, 1L), (partition1, 5L), (partition0, 2L)).map { case (partition, offset) => CommittableConsumerRecord( ConsumerRecord(partition.topic, partition.partition, offset, "key", "value"),