Akka Streams Kafka

Producer

A producer publishes messages to Kafka topics. The message itself contains information about what topic and partition to publish to so you can publish to different topics with the same producer.

The underlaying implementation is using the KafkaProducer, see Javadoc for details.

Producer as a Sink

Producer.plainSink is the easiest way to publish messages. The sink consumes ProducerRecord elements which contains a topic name to which the record is being sent, an optional partition number, and an optional key and value.

Producer as a Sink

Producer.plainSink is the easiest way to publish messages. The sink consumes ProducerRecord elements which contains a topic name to which the record is being sent, an optional partition number, and an optional key and value.

Sharing KafkaProducer

If you have many streams it can be more efficient to share the underlying KafkaProducer.

You can create a KafkaProducer instance from ProducerSettings.

Scala

Java

protected final ProducerSettings<byte[], String> producerSettings = ProducerSettings
  .create(system, new ByteArraySerializer(), new StringSerializer())
  .withBootstrapServers("localhost:9092");
protected final KafkaProducer<byte[], String> kafkaProducer = producerSettings.createKafkaProducer();

The KafkaProducer is passed as a parameter to the Producer factory methods.

Scala

Java

CompletionStage<Done> done =
        Source.range(1, 100)
                .map(n -> n.toString()).map(elem -> new ProducerRecord<byte[], String>("topic1", elem))
                .runWith(Producer.plainSink(producerSettings, kafkaProducer), materializer);