ZIO Kafka Documentation

When consuming records, you must commit the offset of a CommittableRecord to ensure it isn't re-consumed after a crash or restart.

While you can commit offsets individually using record.offset.commit, this is inefficient for high-volume streams. Instead, use ZStream#aggregateAsyncWithin combined with the Consumer.collectOffsets sink to aggregate offsets into an OffsetBatch and commit them in batches at a fixed interval.

consumer
  .plainStream(Subscription.topics(KAFKA_TOPIC), Serde.int, Serde.string)
  .map(_.offset)
  .aggregateAsyncWithin(Consumer.collectOffsets, Schedule.fixed(100.millis))
  .mapZIO(_.commit)
  .runDrain

zio-kafka programs often achieve higher throughput than programs using the standard java-kafka client directly because zio-kafka processes partitions in parallel by default.

Key Performance Notes:

• Parallelism: zio-kafka processes partitions in parallel, whereas the default java-kafka client does not.
• Throughput Threshold: As of late 2024, zio-kafka is estimated to consume faster than the java-kafka client when processing takes more than approximately 1.2ms per 1000 records. The exact threshold depends on various factors.
• Latency vs. Throughput: If your application requires absolute minimum latency and you do not need the ZStream-based API, using the Java-based Kafka client directly may be preferable.

ZIO Kafka handles the conversion between raw byte arrays and high-level Scala types for both Kafka keys and values.

• Producer: Requires a Serializer for both keys and values to convert types to byte arrays.
• Consumer: Requires a Deserializer for both keys and values to convert byte arrays to types.
• Serde: A Serde[K, V] is a combined object that provides both a Serializer and a Deserializer for a specific key and value type.

Common pre-defined serdes are available in the Serdes object, such as Serdes.byteArray, Serdes.string, and Serdes.long.

ZIO Kafka optimizes throughput by grouping records fetched from a broker into chunks. Using operators like mapZIO or tap can break this structure, resulting in a stream where every chunk contains only a single element, significantly reducing performance.

To identify chunk-breaking operators in Scaladoc, look for the warning: This combinator destroys the chunking structure.

Important Note on Execution Order: When moving from chunk-breaking operators to chunk-preserving ones, the evaluation order changes:

• Chunk-breaking (mapZIO(f).mapZIO(g)): f(a), g(a), f(b), g(b)
• Chunk-preserving (Alternatives): f(a), f(b), g(a), g(b)

If g(a) fails in the chunk-preserving model, f(b) will have already been executed. Ensure this change in semantics is acceptable for your application.

Kafka stores records as raw bytes. To work with typed data, ZIO Kafka uses the Serde[R, A] type, which combines a Serializer and a Deserializer for type A within an environment R.

trait Serde[-R, A] {
  def deserialize(data: Array[Byte]): RIO[R, A]
  def serialize(value: A)           : RIO[R, Array[Byte]]
}

The Serde companion object provides built-in serializers/deserializers for primitive types:

• Serde.long
• Serde.int
• Serde.short
• Serde.float
• Serde.double
• Serde.boolean
• Serde.string
• Serde.byteArray
• Serde.byteBuffer
• Serde.uuid

Producer benchmarks test various patterns of record production. Common patterns include:

• Batch Production: Producing batches of records (e.g., 30 batches of 500 small records).
  • produceChunkSeq: Sequential production.
  • produceChunkSeqAsync: Sequential production where the client does not wait for acknowledgements.
  • produceChunkPar: Production using 4 concurrent fibers.
• Single Record Production: Producing individual records (e.g., 100 small records) with lingering disabled.
  • produceSingleRecordSeq: Sequential production.
  • produceSingleRecordSeqAsync: Sequential production without waiting for acknowledgements.
  • produceSingleRecordPar: Production using 4 concurrent fibers.

When consuming from a large number of partitions, the heap usage increases because each partition maintains a record queue.

A rough estimate for required heap is: average record size * number of partitions * max(partitionPreFetchBufferLimit, max.poll.records)

To reduce memory pressure:

1. Decrease partitionPreFetchBufferLimit or max.poll.records.
2. Use a custom FetchStrategy, such as ManyPartitionsQueueSizeBasedFetchStrategy (available since version 2.8.1).

The restartStreamOnRebalancing mode is no longer supported in zio-kafka 3. This mode previously ended all streams during a rebalance, even if their specific partitions were not revoked.

In zio-kafka 3, transactional consuming no longer requires this mode. If your goal is to prevent duplicate processing, use rebalanceSafeCommits instead.

The consumer benchmarks in this project are designed to represent a 'worst-case' scenario for zio-kafka to ensure fair comparison with standard Kafka clients. Because these benchmarks only count received records without any processing, they do not leverage zio-kafka's ability to process records in parallel, which can make the throughput appear lower than in real-world usage.

All consumer benchmarks send approximately 50,000 records of 512 bytes each per run.

Key benchmark types:

• throughput: Uses plainStream with a topic subscription; offsets are not committed.
• throughputWithCommits: Uses plainStream with a topic subscription; offsets are committed.
• manual variants: Use partition assignment instead of topic subscription.