Consumer tracing configurator

Choose tracing options, inspect the spans, and copy the Scala example. Disabling spans keeps processing and commits enabled.

Choose one or three messages in one partition. There is no active application span; the consume helpers keep a receive span active while processing when receive tracing is enabled.

Configure your consumer

Trace timeline

Expand a span to see its attributes and parent. Follow a link to see the message context.

Incoming message contexts

Consumer spans retain links to these contexts in both relationship modes. With ParentFromRecord, single-record and singleton batch process spans also use the corresponding context as parent, continuing its trace.

Generated Scala

Supply a configured otel4s TracerProvider. Run Kafka at localhost:9092 and create the selected topic.

For complete handler examples and span diagrams, see Consumers. For shared tracing options, see Configuration.

Choose a consumer API

API	Handler calls	Process spans	Commits
`consumeRecordTraced`	Once per record	One per record	After all handlers in the chunk succeed
`consumeChunkTraced`	Once per chunk	One for the entire chunk	After the chunk handler succeeds
`recordsTraced`	Once per record	One per record	You choose a downstream commit pipe

The consume helpers keep receive open during processing, then commit after it ends. A failed or canceled handler skips the commit. With recordsTraced, receive ends before processing.

The stream example uses consumer.spans.commitBatchWithin for traced commits. Disabling commit spans switches to the standard fs2-kafka pipe. Both pipes commit offsets.

Configure spans independently

Return None from a setup function to disable a span. For example, disable the batch process span:

import fs2.kafka.otel4s.trace.KafkaTracer

val manualRecordConfig =
  KafkaTracer.Config
    .default
    .withBatchProcessSpanSetup(_ => None)

Configuration	Affects
`withReceiveSpanSetup` / `withoutReceiveSpans`	Receive spans
`withProcessSpanSetup`	Single-record process spans
`withBatchProcessSpanSetup`	Batch process spans
`withoutProcessSpans`	Both single-record and batch process spans
`withCommitSpanSetup` / `withoutCommitSpans`	Commit spans

KafkaTracer.noop disables both spans and propagation.

Choose a process relationship

Link creates linked roots. ParentFromRecord continues the message trace for single-record processing and singleton batches, retaining message links. Missing or invalid headers produce a root; multiple-record batches remain linked roots.

import fs2.kafka.otel4s.trace.ProcessSpanRelationship

val parentedProcessingConfig =
  KafkaTracer.Config.default
    .withProcessSpanSetup { ctx =>
      KafkaTracer.Config.Defaults.processSpanSetup(ctx)
        .map(_.withRelationship(ProcessSpanRelationship.ParentFromRecord))
    }

Set the relationship on the setup returned by withProcessSpanSetup, or independently by withBatchProcessSpanSetup for chunk processing. The setup can choose a relationship per topic or other metadata.

Parented spans also link to an active receive span. With recordsTraced, receive has already ended. Receive and commit behavior stays the same.

Try ParentFromRecord with consumeChunkTraced, then switch between one and three messages to compare parenting.

Add child spans around selected work

Enable “Application span per message” with any consumer API to wrap each handler in an application span. With record APIs, each application span is a child of its library process span. With the chunk API, application spans are children of the batch process span. They inherit its trace, including when the process span uses ParentFromRecord.

A chunk handler can also select records before processing. This example keeps the latest record per key and adds an application span for each selected record:

import cats.effect.IO
import cats.syntax.all._
import fs2.Chunk
import fs2.kafka.ConsumerRecord
import fs2.kafka.consumer.KafkaConsumeChunk.CommitNow
import fs2.kafka.otel4s.trace.TracedKafkaConsumer
import org.typelevel.otel4s.trace.Tracer

def consumeSelected(
    consumer: TracedKafkaConsumer[IO, String, String],
    selectLatest: Chunk[ConsumerRecord[String, String]] => Chunk[ConsumerRecord[String, String]],
    handle: ConsumerRecord[String, String] => IO[Unit]
)(implicit tracer: Tracer[IO]): IO[Nothing] =
  consumer.consumeChunkTraced { chunk =>
    selectLatest(chunk).traverse_ { record =>
      Tracer[IO].span("handle order").surround(handle(record))
    }.as(CommitNow)
  }

The demo processes every message in the selected chunk. Selecting fewer records in a chunk handler creates fewer child spans. The record API examples wrap handlers in the same way:

def consumeWithChildren(
    consumer: TracedKafkaConsumer[IO, String, String],
    handle: ConsumerRecord[String, String] => IO[Unit]
)(implicit tracer: Tracer[IO]): IO[Nothing] =
  consumer.consumeRecordTraced { record =>
    tracer.span("handle order").surround(handle(record))
  }

def recordsWithChildren(
    consumer: TracedKafkaConsumer[IO, String, String],
    handle: ConsumerRecord[String, String] => IO[Unit]
)(implicit tracer: Tracer[IO]): fs2.Stream[IO, fs2.kafka.CommittableOffset[IO]] =
  consumer.recordsTraced { committable =>
    tracer.span("handle order").surround(handle(committable.record)).as(committable.offset)
  }

consumer.spans.process(record)(effect) follows the configured process relationship and retains its message link. It uses withProcessSpanSetup. The chunk handler still commits the original chunk after success.

Read the relationships

Span	Parent in this example	Links
Receive	None	All recognized contexts in the chunk
Record process	None with `Link`; message context with `ParentFromRecord`	Its message context, plus the ambient receive span when parented inside it
Batch process	None, except singleton batches with `ParentFromRecord`	All recognized contexts, plus the ambient receive span when parented inside it
Application child	Record or batch process; otherwise the active receive span, if any	None
Commit	None	None

If process tracing is disabled, application spans are children of receive in the consume helpers when receive tracing is enabled. With recordsTraced, receive has already ended, so application spans become roots. If both process and receive tracing are disabled, application spans are roots. Turning off incoming headers removes message links and makes process spans roots in both modes. It does not change the generated consumer code.

The stream example commits batches of one or three offsets, matching the selected message count. Actual timing and batch sizes depend on your application.

Constant attributes can replace derived attributes. Span names and links stay the same. Selections are saved in the URL. Keep secrets out of shared attribute values.