ZIO Kafka

https://github.com/zio/zio-kafka
Admin
A fast Kafka client for ZIO and ZIO Streams
Tokens:13,672
Snippets:78
Trust Score:7.7
License:Apache-2.0
Update:1 year ago
Show doc for...
Tokens:
Raw
### Implement ZIO Kafka Producer and Consumer in Scala

Source: https://github.com/zio/zio-kafka/blob/master/docs/index.md

This Scala application demonstrates a basic ZIO Kafka streaming setup. It includes a producer that continuously sends random integers to a 'random-topic' and a consumer that reads from the same topic, prints the values to the console, and commits offsets. The application uses `zio-streams` for continuous data flow and `ZIOAppDefault` for execution.

```scala
import zio._
import zio.kafka.consumer._
import zio.kafka.producer.{ Producer, ProducerSettings }
import zio.kafka.serde._
import zio.stream.ZStream

object ReadmeExample extends ZIOAppDefault {

  private val producerRun: ZIO[Any, Throwable, Unit] =
    ZIO.scoped {
      for {
        producer <-
          Producer.make(
            ProducerSettings(List("localhost:9092"))
          )
        _ <- ZStream
          .fromSchedule(Schedule.fixed(2.seconds))
          .mapZIO(_ => Random.nextIntBetween(0, Int.MaxValue))
          .mapZIO { random =>
            producer.produce[Any, Long, String](
              topic = "random-topic",
              key = (random % 4).toLong,
              value = random.toString,
              keySerializer = Serde.long,
              valueSerializer = Serde.string
            )
          }
          .runDrain
      } yield ()
    }

  private val consumerRun: ZIO[Any, Throwable, Unit] =
    ZIO.scoped {
      for {
        consumer <-
          Consumer.make(
            ConsumerSettings(List("localhost:9092"))
              .withGroupId("group")
          )
        _ <- consumer
          .plainStream(Subscription.topics("random"), Serde.long, Serde.string)
          .tap(r => Console.printLine(r.value))
          .map(_.offset)
          .aggregateAsync(Consumer.offsetBatches)
          .mapZIO(_.commit)
          .runDrain
      } yield ()
    }

  override def run: ZIO[Any, Throwable, Unit] =
    ZIO.raceFirst(producerRun, List(consumerRun))
}
```

--------------------------------

### Add ZIO Kafka Dependencies to build.sbt

Source: https://github.com/zio/zio-kafka/blob/master/docs/index.md

To begin using ZIO Kafka, add the necessary library dependencies to your `build.sbt` file. This includes the core `zio-kafka` library and the `zio-kafka-testkit` for testing purposes. For Scala 3 projects using `zio-kafka-testkit`, an additional exclusion might be required to prevent dependency conflicts.

```scala
libraryDependencies += "dev.zio" %% "zio-kafka"         % "@VERSION@"
libraryDependencies += "dev.zio" %% "zio-kafka-testkit" % "@VERSION@" % Test
```

```scala
excludeDependencies += "org.scala-lang.modules" % "scala-collection-compat_2.13"
```

--------------------------------

### Migrate ZIO-Kafka Testkit Admin Client Usage

Source: https://github.com/zio/zio-kafka/blob/master/docs/migrating-to-zio-kafka-3.md

This example demonstrates how to replace deprecated `withAdmin` methods with the new `KafkaTestUtils.makeAdminClient`. The new pattern uses a `for` comprehension to acquire and utilize the admin client, aligning with ZIO's resource management.

```scala
// Old
KafkaTestUtils.withAdmin { adminClient =>
  adminClient.method() // Use the admin client...
}

// New
for {
  adminClient <- KafkaTestUtils.makeAdminClient
  _           <- adminClient.method() // Use the admin client...
} yield ()
```

--------------------------------

### Set up Kafka Broker with Docker Compose

Source: https://github.com/zio/zio-kafka/blob/master/docs/index.md

This Docker Compose configuration defines a single Kafka broker instance. It exposes port 9092 for client connections and sets up necessary environment variables for Kafka to run as a standalone broker and controller, suitable for local development.

```docker-compose
services:
  broker:
    image: apache/kafka:3.9.0
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_BROKER_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
```

--------------------------------

### Provide Consumer Settings and Consumer as Layers (zio-kafka 3.x)

Source: https://github.com/zio/zio-kafka/blob/master/docs/creating-a-consumer.md

For zio-kafka version 3.x, this code demonstrates how to provide both `ConsumerSettings` and the `Consumer` itself as ZIO layers. It utilizes `Consumer.live` which takes the consumer settings as a layer, simplifying the dependency injection setup in the main program's `provide` section.

```scala
.provide(
  consumerSettingsLayer,
  Consumer.live
)
```

--------------------------------

### Minimal ZIO Kafka Producer Test Example

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

This Scala snippet demonstrates a basic test for a Kafka producer using `zio-kafka-testkit` and `zio-test`. It illustrates how to create a producer, send a record, and assert completion, while providing an embedded Kafka instance shared across the entire test suite for efficiency.

```scala
import org.apache.kafka.clients.producer.ProducerRecord
import zio._
import zio.kafka.serde.Serde
import zio.kafka.testkit.Kafka
import zio.kafka.testkit.KafkaTestUtils
import zio.test.TestAspect.{timeout, withLiveClock}
import zio.test._

object ProducerSpec extends ZIOSpecDefault {
  override def spec: Spec[TestEnvironment & Scope, Any] =
    (
      suite("Producer test suite")(
        test("minimal example") {
          for {
            producer <- KafkaTestUtils.makeProducer // (1)
            _ <- producer.produce(new ProducerRecord("topic", "boo", "baa"), Serde.string, Serde.string) // (3)
          } yield assertCompletes // (3)
        }
        // ... more tests ...
      )
        .provideSomeShared[Scope](Kafka.embedded) // (4) Provide an embedded Kafka instance, shared in the entire suite
      ) @@ withLiveClock @@ timeout(2.minutes) // (5)
}
```

--------------------------------

### Test ZIO Kafka Consumer with Embedded Kafka

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

Demonstrates a minimal integration test for a ZIO Kafka consumer. It involves creating a custom topic, producing messages, consuming them, and asserting that the consumed messages match the produced ones. This example utilizes `KafkaTestUtils` for producer/consumer setup and `Kafka.embedded` for providing an in-memory Kafka instance.

```scala
import zio._
import zio.kafka.consumer.Subscription
import zio.kafka.serde.Serde
import zio.kafka.testkit.KafkaTestUtils
import zio.kafka.testkit._
import zio.test.Assertion.hasSameElements
import zio.test.TestAspect.{timeout, withLiveClock}
import zio.test._

object ConsumerSpec extends ZIOSpecDefault {

  override def spec: Spec[TestEnvironment & Scope, Any] =
    (
      suite("Consumer test suite")(
        test("minimal example") {
          // Records (as key/value pairs) to produce and consume
          val kvs: List[(String, String)] = (1 to 5).toList.map(i => (s"key-$i", s"msg-$i"))
          for {
            topic <- Random.nextUUID.map("topic-" + _.toString) // (1)
            client <- Random.nextUUID.map("client-" + _.toString)
            group <- Random.nextUUID.map("group-" + _.toString)

            _ <- KafkaTestUtils.createCustomTopic(topic, partitionCount = 3) // (2)

            producer <- KafkaTestUtils.makeProducer
            _ <- KafkaTestUtils.produceMany(producer, topic, kvs) // (3)

            consumer <- KafkaTestUtils.makeConsumer(clientId = client, groupId = Some(group)) // (4)
            records <- consumer
              .plainStream(Subscription.topics(topic), Serde.string, Serde.string)
              .take(5)
              .runCollect // (5)
            consumed = records.map(r => (r.record.key, r.record.value)).toList
          } yield assert(consumed)(hasSameElements(kvs))
        }
      )
        .provideSomeShared[Scope](Kafka.embedded) // (6)
      ) @@ withLiveClock @@ timeout(2.minutes)
}
```

--------------------------------

### Migrate ZIO-Kafka Consumer with Custom Diagnostics

Source: https://github.com/zio/zio-kafka/blob/master/docs/migrating-to-zio-kafka-3.md

This example illustrates the updated approach for providing custom diagnostics to the ZIO-Kafka consumer. In 3.x, diagnostics are now configured directly on `ConsumerSettings` using `withDiagnostics` instead of being passed as a separate argument to `Consumer.make`.

```scala
// zio-kafka 2.x
val diagnostics = ???
val settings = ConsumerSettings(bootstrap)
val consumer = Consumer.make(settings, diagnostics)

// zio-kafka 3.x
val diagnostics = ???
val settings = ConsumerSettings(bootstrap).withDiagnostics(diagnostics)
val consumer = Consumer.make(settings)
```

--------------------------------

### ZIO Kafka Consumer-Producer with Manual Offset Commit

Source: https://github.com/zio/zio-kafka/blob/master/docs/example-of-consuming-producing-and-committing-offsets.md

This Scala code demonstrates a ZIO Kafka application that consumes messages from an input topic, transforms their values, produces the transformed messages to an output topic, and then manually commits the consumer offsets. It highlights the use of `mapChunksZIO` for efficient batch processing and explicitly manages offset commits, as ZIO Kafka does not support automatic offset committing.

```scala
import zio.ZLayer
import zio.kafka.consumer._
import zio.kafka.producer._
import zio.kafka.serde._
import org.apache.kafka.clients.producer.ProducerRecord

val consumerSettings: ConsumerSettings =
  ConsumerSettings(List("localhost:9092")).withGroupId("group")
val producerSettings: ProducerSettings =
  ProducerSettings(List("localhost:9092"))

for {
  consumer <- Consumer.make(consumerSettings)
  producer <- Producer.make(producerSettings, Serde.int, Serde.string)
  _ <-
    consumer
      .plainStream(Subscription.topics("my-input-topic"), Serde.int, Serde.long)
      .map { record =>
        val key: Int = record.record.key()
        val value: Long = record.record.value()
        val newValue: String = value.toString

        val producerRecord: ProducerRecord[Int, String] =
          new ProducerRecord("my-output-topic", key, newValue)
        (producerRecord, record.offset)
      }
      .mapChunksZIO { chunk =>
        val records = chunk.map(_._1)
        val offsetBatch = OffsetBatch(chunk.map(_._2).toSeq)

        producer.produceChunk[Any, Int, String](records) *>
          offsetBatch.commit.as(Chunk(()))
      }
      .runDrain
} yield ()
```

--------------------------------

### Manage Consumer Lifecycle with ZIO.scoped

Source: https://github.com/zio/zio-kafka/blob/master/docs/creating-a-consumer.md

This example illustrates how to explicitly manage the lifecycle of a `zio-kafka` consumer using `ZIO.scoped`. By wrapping consumer usage within a `ZIO.scoped` block, the consumer's connection to the Kafka cluster is automatically closed when the scope exits, ensuring proper resource cleanup.

```scala
ZIO.scoped {
  // make the consumer
  // use the consumer
}
// end of scope, consumer has closed connection to Kafka cluster.
```

--------------------------------

### Complete ZIO Kafka Transactional Producer Application

Source: https://github.com/zio/zio-kafka/blob/master/docs/transactions.md

This comprehensive example demonstrates a full ZIO application that consumes messages from one Kafka topic, transforms them, and then produces them transactionally to another topic. It sets up consumer and transactional producer settings, handles stream processing with `mapChunksZIO` for transactional operations, and manages ZIO layers for dependency injection.

```scala
import org.apache.kafka.clients.producer.ProducerRecord
import zio._
import zio.kafka.consumer._
import zio.kafka.producer._
import zio.kafka.serde.Serde

import java.util.UUID

object Transactional extends ZIOAppDefault {

  private val topic = "transactional-test-topic"

  val bootstrapBrokers = List("localhost:9092")

  val consumerSettings = ZLayer.succeed {
    ConsumerSettings(bootstrapBrokers)
      .withGroupId("transactional-example-app")
      .withRebalanceSafeCommits(true) // required for transactional producing
      .withMaxRebalanceDuration(30.seconds)
  }

  val producerSettings = ZLayer.succeed {
    TransactionalProducerSettings(
      bootstrapServers = bootstrapBrokers,
      transactionalId = UUID.randomUUID().toString
    )
  }

  private val runConsumerStream: ZIO[Consumer & TransactionalProducer, Throwable, Unit] =
    for {
      consumer              <- ZIO.service[Consumer]
      transactionalProducer <- ZIO.service[TransactionalProducer]
      _                     <- ZIO.logInfo(s"Consuming messages from topic $topic...")
      _ <- consumer
        .plainStream(Subscription.topics(topic), Serde.int, Serde.long)
        .mapChunks { records: Chunk[CommittableRecord[Int, Long]] =>
          records.map { record =>
            val key: Int         = record.record.key()
            val value: Long      = record.record.value()
            val newValue: String = "Copy of " + value.toString

            val producerRecord: ProducerRecord[Int, String] =
              new ProducerRecord("my-output-topic", key, newValue)
            (producerRecord, record.offset)
          }
        }
        .mapChunksZIO { recordsAndOffsets: Chunk[(ProducerRecord[Int, String], Offset)] =>
          ZIO.scoped {
            for {
              _  <- ZIO.addFinalizer(ZIO.logInfo("Completing transaction"))
              tx <- transactionalProducer.createTransaction
              _ <- {
                val (records, offsets) = recordsAndOffsets.unzip
                tx.produceChunkBatch(
                  records,
                  Serde.int,
                  Serde.string,
                  OffsetBatch(offsets)
                )
              }
            } yield Chunk.empty
          }.uninterruptible
        }
        .runDrain
    } yield ()

  override def run: ZIO[Scope, Any, Any] =
    runConsumerStream
      .provide(
        consumerSettings,
        ZLayer.succeed(Consumer.NoDiagnostics),  // No longer needed in zio-kafka 3.x
        Consumer.live,
        producerSettings,
        TransactionalProducer.live
      )

}
```

--------------------------------

### Share Kafka Instance Across ZIO Test Suites with ZIOSpecWithKafka

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

This snippet demonstrates how to use the `ZIOSpecWithKafka` trait to share a single Kafka instance across multiple ZIO test suites. By extending this trait, tests in different files can reuse the same Kafka environment, significantly reducing setup time and improving overall test execution speed. It's particularly useful for large test suites where Kafka initialization is a bottleneck.

```scala
// In `src/test/scala/io/example/producer/ProducerSpec.scala`
object ProducerSpec extends ZIOSpecWithKafka { // Note `ZIOSpecWithKafka`
  override def spec: Spec[TestEnvironment & Scope & Kafka, Any] =
    suite("Producer test suite")(
      // ... tests ...
    ) @@ timeout(2.minutes)
}

// In `src/test/scala/io/example/consumer/ConsumerSpec.scala`
object ConsumerSpec extends ZIOSpecWithKafka { // Note `ZIOSpecWithKafka`
  override def spec: Spec[TestEnvironment & Scope & Kafka, Any] =
    suite("Consumer test suite")(
      // ... tests ...
    ) @@ timeout(2.minutes)
}
```

--------------------------------

### Kafka Consumer Offset Retrieval Strategies in zio-kafka

Source: https://github.com/zio/zio-kafka/blob/master/docs/partition-assignment-and-offset-retrieval.md

This section outlines the different strategies `zio-kafka` offers for managing consumer offsets, including automatic retrieval from Kafka (starting at latest or earliest message) and manual/external offset storage, useful for consistency requirements.

```APIDOC
OffsetRetrieval.Auto()
  Description: Offsets in Kafka, start at latest message if no offset committed.
```

```APIDOC
OffsetRetrieval.Auto(AutoOffsetStrategy.Earliest)
  Description: Offsets in Kafka, start at earliest message if no offset committed.
```

```APIDOC
Manual(getOffsets: Set[TopicPartition] => Task[Map[TopicPartition, Long]])
  Description: Manual/external offset storage. The 'getOffsets' function is called for each topic-partition assigned to the consumer.
```

--------------------------------

### Manual Graceful Kafka Consumer Shutdown with StreamControl

Source: https://github.com/zio/zio-kafka/blob/master/docs/consuming-kafka-topics-using-zio-streams.md

This example illustrates how to manually control the graceful shutdown of a Kafka consumer stream using the `StreamControl` object. It allows stopping the fetching of new records with `control.end` and then waiting for the existing stream processing to complete by joining the stream's fiber, providing fine-grained control over termination conditions.

```scala
import zio.Console.printLine
import zio.kafka.consumer._

ZIO.scoped {
    for {
      control <- consumer.partitionedStreamWithControl(
          Subscription.topics("topic150"),
          Serde.string,
          Serde.string
        )
      fiber <- control.stream.flatMapPar(Int.MaxValue) { case (topicPartition, partitionStream) =>
            partitionStream
              .tap(record => printLine(s"key: ${record.key}, value: ${record.value}"))
              .map(_.offset)
          }
          .aggregateAsync(Consumer.offsetBatches)
          .mapZIO(_.commit)
          .runDrain
          .forkScoped
      // At some point in your code, when some condition is met
      _ <- control.end // Stop fetching more records
      _ <- fiber.join // Wait for graceful completion of the stream
    } yield ()
}
```

--------------------------------

### Generate Random Kafka Test Values with KafkaRandom Trait

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

This example illustrates the usage of the `KafkaRandom` trait, which provides utility methods for generating random Kafka-related values such as topics, client IDs, and group IDs. To use `KafkaRandom`, it must be mixed into your test suite, and the `kafkaPrefix` method needs to be overridden to provide a unique prefix for the generated values, ensuring isolation between test suites.

```scala
import zio.kafka.testkit.Kafka
import zio.kafka.testkit.KafkaRandom
import zio.kafka.testkit.KafkaTestUtils
import zio.test._
import zio._

object MyServiceSpec extends ZIOSpecDefault with KafkaRandom {
  // Required when mixing in the `KafkaRandom` trait
  // The best is to use a different prefix for each test suite
  override def kafkaPrefix: String = "my-service"

  override def spec: Spec[TestEnvironment & Scope, Any] =
    suite("MyService")(
      test("minimal example") {
        for {
          topic    <- randomTopic  // Comes from `KafkaRandom`
          clientId <- randomClient // Comes from `KafkaRandom`
          groupId  <- randomGroup  // Comes from `KafkaRandom`
          // ... 
        } yield assertCompletes
      }
    ).provideSomeShared[Scope](Kafka.embedded)
}
```

--------------------------------

### Initialize Zio-kafka ConsumerSettings

Source: https://github.com/zio/zio-kafka/blob/master/docs/consumer-tuning.md

Demonstrates the basic initialization of `ConsumerSettings` with bootstrap servers, group ID, and additional properties, serving as the foundation for further consumer configuration.

```scala
val settings = ConsumerSettings(bootstrapServers)
  .withGroupId(groupId)
  .withProperties(properties)
  .... etc.
```

--------------------------------

### Initialize zio-kafka Consumer Settings and Instance

Source: https://github.com/zio/zio-kafka/blob/master/docs/creating-a-consumer.md

This snippet demonstrates the fundamental steps to configure a Kafka consumer using `ConsumerSettings` and then create a `Consumer` instance with `Consumer.make`. It highlights the use of `withGroupId` for consumer group assignment and notes that the consumer requires a `Scope` for proper resource management.

```scala
import zio.*
import zio.kafka.consumer.{ Consumer, ConsumerSettings }

val consumerSettings: ConsumerSettings =
  ConsumerSettings(List("localhost:9092")).withGroupId("group")
val consumer: ZIO[Scope, Throwable, Consumer] =
  Consumer.make(consumerSettings)
```

--------------------------------

### Run ZIO Kafka Comparison Benchmarks Locally

Source: https://github.com/zio/zio-kafka/blob/master/zio-kafka-bench/README.md

This command sequence is used to execute a specific set of comparison benchmarks for ZIO Kafka using sbt and JMH. It first cleans the project, then compiles the tests, and finally runs the benchmarks matching the 'comparison' package with detailed JMH configurations for warmup iterations, benchmark iterations, time per iteration, worker threads, forks, and immediate error failure.

```scala
clean
Test/compile
zioKafkaBench/Jmh/run -wi 10 -i 10 -r 1 -w 1 -t 1 -f 5 -foe true .*comparison.*
```

--------------------------------

### Kafka and Zio-kafka Consumer Configuration Parameters Reference

Source: https://github.com/zio/zio-kafka/blob/master/docs/consumer-tuning.md

Comprehensive documentation of key configuration parameters for Kafka consumers and Zio-kafka's `ConsumerSettings`, detailing their purpose, typical values, and impact on consumer performance and resource usage.

```APIDOC
ConsumerSettings Parameters:
  pollTimeout:
    Description: Maximum time to block while polling the Kafka consumer. Zio-kafka's default is 50ms, good for low latency. Set higher (e.g., 500ms) for better throughput.
    Type: Duration
  partitionPreFetchBufferLimit:
    Description: Threshold for pre-fetching records from Kafka into partition queues. When the number of records in a partition queue is at or below this value, zio-kafka will start to pre-fetch. Default is 1024. Can be decreased for large records or high partition counts, or disabled using `withoutPartitionPreFetching`.
    Type: Int

Kafka Consumer Configuration Parameters:
  max.poll.interval.ms:
    Description: The maximum time between polls before a consumer is considered stalled and its partitions are revoked. Default is 5 minutes. Must be substantially higher than the maximum processing time per batch to avoid consumer shutdowns.
    Type: Long
  max.poll.records:
    Description: The maximum number of records a single poll will return. Kafka defaults this to 500. Can be set higher for more throughput, or lower for lower latency.
    Type: Int

Zio-kafka ConsumerSettings Utility Methods:
  tuneForHighThroughput():
    Description: Pre-configures ConsumerSettings for high throughput use cases.
    Returns: ConsumerSettings
  tuneForLowLatency():
    Description: Pre-configures ConsumerSettings for low latency use cases.
    Returns: ConsumerSettings
  withoutPartitionPreFetching():
    Description: Disables pre-fetching of records by setting `partitionPreFetchBufferLimit` to 0.
    Returns: ConsumerSettings
```

--------------------------------

### ZIO-Kafka 2.x Consumer Layer Provisioning

Source: https://github.com/zio/zio-kafka/blob/master/docs/migrating-to-zio-kafka-3.md

Illustrates the standard way of providing consumer settings and diagnostics layers in ZIO-Kafka 2.x applications, where each layer is provided independently to the application.

```scala
.provide(
  consumerSettingsLayer,
  diagnosticsLayer,
  Consumer.live
)
```

--------------------------------

### Provide Consumer Settings and Consumer as Layers (zio-kafka 2.x)

Source: https://github.com/zio/zio-kafka/blob/master/docs/creating-a-consumer.md

This snippet illustrates the layer provision strategy for `zio-kafka` version 2.x, where `Consumer.live` requires both consumer settings and diagnostics settings as layers. It shows how to include `ZLayer.succeed(Consumer.NoDiagnostics)` alongside `consumerSettingsLayer` when providing dependencies.

```scala
.provide(
  consumerSettingsLayer,
  ZLayer.succeed(Consumer.NoDiagnostics),
  Consumer.live
)
```

--------------------------------

### Add zio-kafka-testkit Dependency to build.sbt

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

This snippet shows how to add the `zio-kafka-testkit` library as a test dependency in your `build.sbt` file. Replace `<latest-version>` with the actual version number to ensure you're using the latest compatible version.

```scala
libraryDependencies += "dev.zio" %% "zio-kafka-testkit" % "<latest-version>" % Test
```

--------------------------------

### ZIO-Kafka 3.x Merged Consumer and Diagnostics Layer

Source: https://github.com/zio/zio-kafka/blob/master/docs/migrating-to-zio-kafka-3.md

Demonstrates the ZIO-Kafka 3.x approach to layer provisioning, where `consumerSettingsLayer` and `diagnosticsLayer` are merged into a single `ZLayer`. This new composition applies diagnostics directly to the consumer settings, reflecting changes in the API and promoting cleaner layer management.

```scala
.provide(
  (consumerSettingsLayer ++ diagnosticsLayer) >>> ZLayer {
    for {
      settings <- ZIO.service[ConsumerSettings]
      diagnostics <- ZIO.service[Consumer.ConsumerDiagnostics]
    } yield settings.withDiagnostics(diagnostics)
  },
  Consumer.live
)
```

--------------------------------

### Implement ZIO Service Pattern for Consumer

Source: https://github.com/zio/zio-kafka/blob/master/docs/migrating-to-zio-kafka-3.md

This Scala code demonstrates how to apply the ZIO Service Pattern to integrate `zio-kafka`'s `Consumer` service. It shows how to create a `ZLayer` that provides the `Consumer` to a service class, allowing direct access to the consumer instance rather than relying on deprecated accessor methods. This is the recommended approach for managing services in ZIO applications.

```scala
import zio.kafka.consumer.Consumer

trait Service {
    def someMethod: ZIO[Any, Throwable, String]
}

object Service {
  
  def layer: ZLayer[Consumer, Throwable, Service] = ZLayer {
      for {
        consumer <- ZIO.service[Consumer]
      } yield ServiceLive(consumer)
  }
  
}

case class ServiceLive(consumer: Consumer) extends Service {

  override def someMethod: ZIO[Any, Throwable, String] = {
    // use `consumer`, not `Consumer`:
    consumer.plainStream(/*...*/).runHead
  }
}
```

--------------------------------

### View Kafka Keystore Contents

Source: https://github.com/zio/zio-kafka/blob/master/zio-kafka-testkit/src/main/resources/README.md

This command lists the detailed contents of the `kafka.keystore.jks` file. It displays information such as aliases, owners, and validity periods for each entry within the keystore. The password for viewing is `123456`.

```Shell
keytool -list -v -keystore keystore/kafka.keystore.jks
```

--------------------------------

### Provide zio-kafka Consumer as a ZIO Layer

Source: https://github.com/zio/zio-kafka/blob/master/docs/creating-a-consumer.md

This snippet shows how to construct a `Consumer` as a ZIO layer, making it suitable for applications following the ZIO service pattern. By using `ZLayer.scoped`, the consumer's resources are automatically managed and closed when the layer is no longer required, promoting robust application design.

```scala
import zio.*
import zio.kafka.consumer.{ Consumer, ConsumerSettings }

val consumerLayer: Layer[Any, Throwable, Consumer] =
  ZLayer.scoped { // (1)
    val consumerSettings: ConsumerSettings =
      ConsumerSettings(List("localhost:9092")).withGroupId("group")
    Consumer.make(consumerSettings)
  }
```

--------------------------------

### View Kafka Truststore Contents

Source: https://github.com/zio/zio-kafka/blob/master/zio-kafka-testkit/src/main/resources/README.md

This command lists the detailed contents of the `kafka.truststore.jks` file. It shows information like aliases, owners, and validity periods for the trusted certificates. The password for viewing is `123456`.

```Shell
keytool -list -v -keystore truststore/kafka.truststore.jks
```

--------------------------------

### View CA Private Key Details

Source: https://github.com/zio/zio-kafka/blob/master/zio-kafka-testkit/src/main/resources/README.md

This command uses OpenSSL to display the textual representation of the `ca-key` RSA certificate, which is a private key. It provides details about the key's structure. The password for viewing is `123456`.

```Shell
openssl rsa -in truststore/ca-key -text
```

--------------------------------

### Configure Zio-kafka ConsumerSettings for Throughput or Latency

Source: https://github.com/zio/zio-kafka/blob/master/docs/consumer-tuning.md

Shows how to use predefined utility methods `tuneForHighThroughput` and `tuneForLowLatency` on `ConsumerSettings` to quickly optimize consumer behavior for specific performance goals, abstracting away individual parameter adjustments.

```scala
val highThroughputSettings = ConsumerSettings(bootstrapServers).tuneForHighThroughput
val lowLatencySettings = ConsumerSettings(bootstrapServers).tuneForLowLatency
```

--------------------------------

### APIDOC: Kafka Service Trait Definition

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

Defines the `Kafka` trait, which represents a Kafka instance within tests. It provides methods to retrieve the bootstrap servers for connecting to Kafka and a method to stop the Kafka instance.

```APIDOC
trait Kafka {
  def bootstrapServers: List[String]

  def stop(): UIO[Unit]
}
```

--------------------------------

### Migrate ZIO-Kafka Test Suite with Explicit Producer

Source: https://github.com/zio/zio-kafka/blob/master/docs/migrating-to-zio-kafka-3.md

This snippet illustrates the migration of ZIO-Kafka test suites from using deprecated implicit producer methods to explicitly providing a producer. The new approach requires a `Scope` in the test environment and passes the producer directly to `produceOne`.

```scala
// Old
override def spec: Spec[TestEnvironment, Any] =
  suite("old example suite")(
    test("uses a producer") {
      for {
        _ <- KafkaTestUtils.produceOne("topic", "key", "message")
      } yield assertCompletes
    }
  )
    .provideSome[Kafka](KafkaTestUtils.producer)
    .provideShared(Kafka.embedded)

// New
// Added `Scope`:
override def spec: Spec[TestEnvironment & Scope, Any] =
  suite("new example suite")(
    test("uses a producer") {
      for {
        // Make a producer explicitly (instead of via a layer):
        producer <- KafkaTestUtils.makeProducer
        // Notice explicit producer parameter:
        _ <- KafkaTestUtils.produceOne(producer, "topic", "key", "message")
      } yield assertCompletes
    }
    // Producer layer removed.
    .provideSomeShared[Scope](Kafka.embedded)
    // Using `provideSomeShared[Scope]` instead of `provideShared`
  )
```

--------------------------------

### Kafka Partition Assignment Methods in zio-kafka

Source: https://github.com/zio/zio-kafka/blob/master/docs/partition-assignment-and-offset-retrieval.md

This section details the various methods available in `zio-kafka` for subscribing to Kafka topics and partitions, allowing for automatic assignment based on topics or patterns, or manual assignment to specific partitions.

```APIDOC
Subscription.topics("my_topic", "other_topic")
  Description: One or more topics, automatic partition assignment.
```

```APIDOC
Subscription.pattern("topic.*")
  Description: Topics matching a pattern.
```

```APIDOC
Subscription.manual("my_topic" -> 1, "my_topic" -> 2)
  Description: Manual partition assignment.
```

--------------------------------

### Create Custom Serde for Instant Type

Source: https://github.com/zio/zio-kafka/blob/master/docs/serialization-and-deserialization.md

Demonstrates how to create a custom `Serde` for `java.time.Instant` by transforming an existing `Serdes.long` using the `inmap` method. This allows converting epoch milliseconds to `Instant` and vice-versa.

```scala
import java.time.Instant
import zio.kafka.serde._

val instantSerde: Serde[Any, Instant] =
  Serdes.long.inmap(java.time.Instant.ofEpochMilli)(_.toEpochMilli)
```

--------------------------------

### Add ZIO Kafka Dependencies to build.sbt

Source: https://github.com/zio/zio-kafka/blob/master/README.md

To incorporate ZIO Kafka into your Scala project, add the `zio-kafka` and `zio-kafka-testkit` libraries as dependencies in your `build.sbt` file. The `zio-kafka-testkit` module is typically scoped for testing purposes.

```Scala
libraryDependencies += "dev.zio" %% "zio-kafka"         % "3.0.0"
libraryDependencies += "dev.zio" %% "zio-kafka-testkit" % "3.0.0" % Test
```

--------------------------------

### Set up Local Kafka Broker with Docker Compose

Source: https://github.com/zio/zio-kafka/blob/master/README.md

This Docker Compose configuration defines a single Kafka broker service. It sets up the necessary environment variables for a standalone Kafka instance, including listener configurations and controller roles, making it ready for local development and testing.

```docker-compose
services:
  broker:
    image: apache/kafka:3.9.0
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_BROKER_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
```

--------------------------------

### Share Kafka Instance Across Multiple Test Suites

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

This Scala snippet illustrates how to share a single embedded Kafka instance across different test files (suites) by extending the `ZIOSpecWithKafka` trait. This approach optimizes startup time for large test suites by initializing Kafka only once for multiple test files.

```scala
object ProducerSpec extends ZIOSpecWithKafka { // (1)
  override def spec: Spec[TestEnvironment & Scope & Kafka, Any] =
    (
      suite("Producer test suite")(
        // ... tests ...
      )
        .provideSome[Scope & Kafka](/* ...other layers... */) // (2)
      )
}
```

--------------------------------

### APIDOC: Kafka Companion Object Layers

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

Documents the companion object for the `Kafka` trait, which provides various `ZLayer` instances for creating and managing Kafka instances in tests. These layers include options for embedded in-memory Kafka, connecting to a local Kafka instance, and embedded instances with SASL or SSL authentication.

```APIDOC
object Kafka {
  /**
   * Creates an in-memory Kafka instance with a random port.
   */
  val embedded: ZLayer[Any, Throwable, Kafka]

  /**
   * Will connect to a Kafka instance running on localhost:9092 (with Docker, for example).
   */
  val local: ULayer[Kafka]

  /**
   * Creates an in-memory Kafka instance with a random port and SASL authentication configured.
   */
  val saslEmbedded: ZLayer[Any, Throwable, Kafka.Sasl]

  /**
   * Creates an in-memory Kafka instance with a random port and SSL authentication configured.
   */
  val sslEmbedded: ZLayer[Any, Throwable, Kafka]
}
```

--------------------------------

### Implement ZIO Kafka Producer and Consumer in Scala

Source: https://github.com/zio/zio-kafka/blob/master/README.md

This Scala application, built with ZIO Kafka and ZIO Streams, demonstrates a complete Kafka producer and consumer workflow. The producer continuously generates random numbers and sends them as messages to a Kafka topic, while the consumer reads these messages, prints their values, and manages offset commits. It showcases concurrent execution of producer and consumer logic.

```scala
import zio._
import zio.kafka.consumer._
import zio.kafka.producer.{ Producer, ProducerSettings }
import zio.kafka.serde._
import zio.stream.ZStream

object ReadmeExample extends ZIOAppDefault {

  private val producerRun: ZIO[Any, Throwable, Unit] =
    ZIO.scoped {
      for {
        producer <-
          Producer.make(
            ProducerSettings(List("localhost:9092"))
          )
        _ <- ZStream
          .fromSchedule(Schedule.fixed(2.seconds))
          .mapZIO(_ => Random.nextIntBetween(0, Int.MaxValue))
          .mapZIO { random =>
            producer.produce[Any, Long, String](
              topic = "random-topic",
              key = (random % 4).toLong,
              value = random.toString,
              keySerializer = Serde.long,
              valueSerializer = Serde.string
            )
          }
          .runDrain
      } yield ()
    }

  private val consumerRun: ZIO[Any, Throwable, Unit] =
    ZIO.scoped {
      for {
        consumer <-
          Consumer.make(
            ConsumerSettings(List("localhost:9092"))
              .withGroupId("group")
          )
        _ <- consumer
          .plainStream(Subscription.topics("random"), Serde.long, Serde.string)
          .tap(r => Console.printLine(r.value))
          .map(_.offset)
          .aggregateAsync(Consumer.offsetBatches)
          .mapZIO(_.commit)
          .runDrain
      } yield ()
    }

  override def run: ZIO[Any, Throwable, Unit] =
    ZIO.raceFirst(producerRun, List(consumerRun))
}
```

--------------------------------

### Replace Consumer Accessor with ZIO.service

Source: https://github.com/zio/zio-kafka/blob/master/docs/migrating-to-zio-kafka-3.md

This Scala snippet illustrates an alternative, more direct approach to replace deprecated `Consumer` accessor methods. It shows how to retrieve the `Consumer` instance using `ZIO.service` within a `for` comprehension and then invoke its methods, providing a clear and explicit way to interact with the service without using the ZIO Service Pattern for the entire application.

```scala
for {
  consumer <- ZIO.service[Consumer]
  _        <- consumer.method(...)
} yield ()
```

--------------------------------

### Consume Kafka Topics Sequentially with ZIO plainStream

Source: https://github.com/zio/zio-kafka/blob/master/docs/consuming-kafka-topics-using-zio-streams.md

Demonstrates how to consume messages from Kafka topics sequentially using ZIO Kafka's `plainStream` method. It shows how to subscribe to topics, process `CommittableRecord` objects, extract keys and values, and manage offsets by aggregating them into `OffsetBatch`es for committing. It also highlights a performance consideration regarding stream operators like `tap` and `mapZIO`.

```scala
import zio.Console.printLine
import zio.kafka.consumer._

consumer
  .plainStream(Subscription.topics("topic150"), Serde.string, Serde.string)
  .tap(cr => printLine(s"key: ${cr.record.key}, value: ${cr.record.value}"))
  .map(_.offset)
  .aggregateAsync(Consumer.offsetBatches)
  .mapZIO(_.commit)
  .runDrain
```

--------------------------------

### Share Kafka Producer Instance Across Tests

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

This Scala snippet shows how to share a single `Producer` instance across all tests within a suite by providing the `KafkaTestUtils.producer` layer along with `Kafka.embedded`. This can improve test performance by avoiding repeated producer creation.

```scala
suite("producer test suite")(
  // ... tests ...
).provideSomeShared[Scope](Kafka.embedded, KafkaTestUtils.producer)
```

--------------------------------

### Kafka 4.0.0 AdminClient Method Deprecations and Replacements

Source: https://github.com/zio/zio-kafka/blob/master/docs/migrating-to-zio-kafka-3.md

This section documents the removal of specific `AdminClient` methods in Kafka 4.0.0 and their recommended replacements. `AdminClient.alterConfigs` and `AdminClient.alterConfigsAsync` are no longer available; users should migrate to `AdminClient.incrementalAlterConfigs`.

```APIDOC
Removed Methods:
  - AdminClient.alterConfigs
  - AdminClient.alterConfigsAsync

Replacement Method:
  - AdminClient.incrementalAlterConfigs
```

--------------------------------

### Provide Separate Embedded Kafka Instance Per Test

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

This Scala snippet demonstrates how to configure `zio-kafka-testkit` to provide a new, separate embedded Kafka instance for each individual test within a suite. This is achieved by using `provideSome` instead of `provideSomeShared`, ensuring test isolation but potentially increasing test execution time.

```scala
suite("producer test suite")(
  // ... tests ...
).provideSome[Scope](Kafka.embedded)
```

--------------------------------

### Run ZIO Kafka Tests Sequentially

Source: https://github.com/zio/zio-kafka/blob/master/docs/writing-tests.md

This Scala snippet shows how to ensure tests run sequentially using `TestAspect.sequential`. This is useful to prevent interference when tests share Kafka resources and cannot use unique topics, ensuring predictable test outcomes.

```scala
import zio.test.TestAspect.sequential

suite("test suite")(
 // ... tests ...
) @@ sequential
```

--------------------------------

### Create ZIO Kafka Consumer Layer

Source: https://github.com/zio/zio-kafka/blob/master/docs/transactions.md

Defines a ZIO layer for the Kafka Consumer, configured with rebalance-safe commits and a specific group ID. It sets up the consumer settings including bootstrap brokers and a reduced max rebalance duration for better control.

```scala
val bootstrapBrokers = List("localhost:9092")

val consumerLayer: ZLayer[Any, Throwable, Consumer] = ZLayer {
  val consumerSettings = ConsumerSettings(bootstrapBrokers)
    .withGroupId("somegroupid")
    .withRebalanceSafeCommits(true) // required!
    .withMaxRebalanceDuration(30.seconds)
  Consumer.make(consumerSettings)
}
```

--------------------------------

### Placeholder for Batch Production of Records

Source: https://github.com/zio/zio-kafka/blob/master/docs/transactions.md

A placeholder snippet demonstrating the use of `mapChunksZIO` for processing records in batches. This approach is crucial for optimizing throughput by preserving the underlying chunking structure of the consumer stream, unlike `mapZIO`.

```scala
producedRecordsStream
  .mapChunksZIO { recordsAndOffsets: Chunk[(ProducerRecord[Int, String], Offset)] =>
    ???
  }
```

--------------------------------

### Handle Deserialization Failures with Serde.asTry

Source: https://github.com/zio/zio-kafka/blob/master/docs/serialization-and-deserialization.md

Illustrates how to use `Serde.asTry` to wrap deserialization results in a `scala.util.Try`, allowing graceful handling of failures within a Kafka consumer stream. Records that fail deserialization can be skipped or logged, and their offsets committed to maintain stream progress.

```scala
import zio._, stream._
import zio.kafka.consumer._
import zio.kafka.serde._
import scala.util.{Try, Success, Failure}

val keySerde = Serdes.string
val valueSerde = Serdes.string.asTry   // <-- using `.asTry`

Consumer.make(consumerSettings).flatMap { consumer =>
  consumer
    .plainStream(Subscription.topics("topic150"), keySerde, valueSerde)
    .mapZIO { record => // ⚠️ see section about `mapZIO` below!
      val tryValue: Try[String] = record.record.value()
      val offset: Offset = record.offset

      tryValue match {
        case Success(value) =>
          // Action for successful deserialization
          someEffect(value).as(offset)
        case Failure(exception) =>
          // Possibly log the exception or take alternative action
          ZIO.succeed(offset)
      }
    }
    .aggregateAsync(Consumer.offsetBatches)
    .mapZIO(_.commit)
    .runDrain
}
```

--------------------------------

### Consume Kafka Topics in Parallel with ZIO partitionedStream

Source: https://github.com/zio/zio-kafka/blob/master/docs/consuming-kafka-topics-using-zio-streams.md

Illustrates how to consume Kafka topics with increased parallelism by processing partitions concurrently using ZIO Kafka's `partitionedStream`. This method creates a nested stream of partitions, allowing each partition to be processed on a separate fiber. It explains the use of `flatMapPar` with `Int.MaxValue` for optimal parallelism and how offsets from all sub-streams are merged into a single stream for unified committing.

```scala
import zio.Console.printLine
import zio.kafka.consumer._

consumer
  .partitionedStream(Subscription.topics("topic150"), Serde.string, Serde.string)
  .flatMapPar(Int.MaxValue) { case (topicPartition, partitionStream) =>
    ZStream.fromZIO(
      printLine(s"Starting stream for topic '${topicPartition.topic}' partition ${topicPartition.partition}")
    ) *>
      partitionStream
        .tap(record => printLine(s"key: ${record.key}, value: ${record.value}"))
        .map(_.offset)
  }
  .aggregateAsync(Consumer.offsetBatches)
  .mapZIO(_.commit)
  .runDrain
```

--------------------------------

### Accessing Java Kafka Client Metrics

Source: https://github.com/zio/zio-kafka/blob/master/docs/metrics.md

Describes how to access the raw metrics from the underlying Java Kafka consumer and producer clients using Zio-Kafka's API. These methods provide a live view of the internal metrics.

```APIDOC
Consumer.metrics
Producer.metrics
```

--------------------------------

### Perform Deserialization Directly in Consumer Stream

Source: https://github.com/zio/zio-kafka/blob/master/docs/serialization-and-deserialization.md

Shows how to defer deserialization to the consumer stream by using `Serdes.byteArray` as a pass-through serde. This approach offers more freedom and control over error handling, allowing for custom recovery logic and potential optimizations like operating on chunks of records.

```scala
import zio._, stream._
import zio.kafka.consumer._

def deserialize(value: Array[Byte]): ZIO[Any, Throwable, Message] = ???

Consumer.make(consumerSettings).flatMap { consumer =>
  consumer
    .plainStream(Subscription.topics("topic150"), Serde.byteArray, Serde.byteArray)
    .mapZIO { record => // ⚠️ see section about `mapZIO` below!
      val value: Array[Byte] = record.record.value()
      val offset: Offset = record.offset

      deserialize(value)
        // possible action to take if deserialization fails:
        .recoverWith(_ => someEffect(value))
        .flatMap(processMessage)
        .as(offset)
    }
    .aggregateAsync(Consumer.offsetBatches)
    .mapZIO(_.commit)
    .runDrain
}
```

--------------------------------

### Share ZIO Kafka Consumer Across Multiple Topic Streams in Scala

Source: https://github.com/zio/zio-kafka/blob/master/docs/sharing-consumer.md

This Scala code demonstrates how to initialize and share a single `zio-kafka` Consumer instance across multiple Kafka topic streams. It sets up a Consumer with common settings, then defines two separate streams (`stream1` for 'topic1' and `stream2` for 'topic2'), each with its own deserializers and processing logic. Both streams utilize the same shared Consumer, which is provided via a ZLayer, and are run concurrently using `zipPar`.

```scala
import zio._
import zio.Console.printLine
import zio.kafka.consumer._

// Create a single Consumer instance
val consumerSettings: ConsumerSettings = ConsumerSettings(List("localhost:9092")).withGroupId("group")
val consumer: ZLayer[Any, Throwable, Consumer] = ZLayer.scoped(Consumer.make(consumerSettings))

// Define two streams, each for a different topic and key-value types
val stream1 = ZIO.serviceWithZIO[Consumer] { consumer =>
  consumer.plainStream(Subscription.topics("topic1"), Serde.string, Serde.string)
    .tap(cr => printLine(s"For topic1, got key: ${cr.record.key}, value: ${cr.record.value}"))
    .map(_.offset)
    .aggregateAsync(Consumer.offsetBatches)
    .mapZIO(_.commit)
    .runDrain
}

val stream2 = ZIO.serviceWithZIO[Consumer] { consumer =>
  consumer.plainStream(Subscription.topics("topic2"), Serde.uuid, Serde.int)
    .tap(cr => printLine(s"For topic2, got key: ${cr.record.key}, value: ${cr.record.value}"))
    .map(_.offset)
    .aggregateAsync(Consumer.offsetBatches)
    .mapZIO(_.commit)
    .runDrain
}

// Run both streams with the same consumer instance
(stream1 zipPar stream2).provideSomeLayer(consumer)
```

--------------------------------

### Set Kafka Consumer Max Poll Interval using Zio-kafka (Newer Versions)

Source: https://github.com/zio/zio-kafka/blob/master/docs/consumer-tuning.md

Illustrates how to configure the `max.poll.interval.ms` property using the `withMaxPollInterval` method in Zio-kafka's `ConsumerSettings` for versions 2.5.1 and later, simplifying the setting of this critical timeout.

```scala
  .withMaxPollInterval(15.minutes)
```
Tokens:
Raw