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.

Settings

When creating a producer stream you need to pass in ProducerSettings that defines things like:

• bootstrap servers of the Kafka cluster
• serializers for the keys and values
• tuning parameters

NOTE:

Specifying null as a key/value serializer uses default serializer for key/value type. Built-in serializers are available in Confluent.Kafka.Serializers class.

By default when creating ProducerSettings with the ActorSystem parameter it uses the config section akka.kafka.producer.

Defining Kafka Properties Directly Inside HOCON

You can embed Kafka properties directly inside the HOCON configuration by declaring them inside the kafka-clients section:

Importing Confluent.Kafka.ProducerConfig Directly Into ProducerSettings

Working with ProducerConfig is a lot more convenient than having to use the ProducerSettings.WithProperty method because you don't have to memorize all of Kafka property names. You can import ProducerConfig directly into Akka.Streams.Kafka.ProducerSettings by using the convenience method ProducerSettings.WithProducerConfig to import all of the defined Kafka properties.

Default Producer HOCON Settings

PlainSink

KafkaProducer.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 a value.

The materialized value of the sink is a Task which is completed with result when the stream completes or with exception if an error occurs.

Producer as a Flow

Sometimes there is a need for publishing messages in the middle of the stream processing, not as the last step, and then you can use KafkaProducer.FlexiFlow.

This flow accepts implementations of Akka.Streams.Kafka.Messages.IEnvelope and return Akka.Streams.Kafka.Messages.IResults elements. IEnvelope elements contain an extra field to pass through data, the so called passThrough. Its value is passed through the flow and becomes available in the ProducerMessage.Results's PassThrough. It can for example hold a Akka.Streams.Kafka.Messages.CommittableOffset or Akka.Streams.Kafka.Messages.CommittableOffsetBatch (from a KafkaConsumer.CommittableSource) that can be committed after publishing to Kafka:

Produce a single message to Kafka

To create one message to a Kafka topic, use the Akka.Streams.Kafka.Messages.Message implementation of IEnvelop.

It can be created with ProducerMessage.Single helper:

The flow with ProducerMessage.Message will continue as ProducerMessage.Result elements containing:

• the original input Message
• the record metadata, and
• access to PassThrough within the message

Let one stream element produce multiple messages to Kafka

The ProducerMessage.MultiMessage implementation of IEnvelope contains a list of ProducerRecords to produce multiple messages to Kafka topics:

The flow with ProducerMessage.MultiMessage will continue as ProducerMessage.MultiResult elements containing:

• a list of MultiResultPart with
  • the original input message
  • the record metadata
• the PassThrough data

Let a stream element pass through, without producing a message to Kafka

The ProducerMessage.PassThroughMessage allows to let an element pass through a Kafka flow without producing a new message to a Kafka topic. This is primarily useful with Kafka commit offsets and transactions, so that these can be committed without producing new messages.

For flows the ProducerMessage.PassThroughMessages continue as ProducerMessage.PassThroughResult elements containing the passThrough data.

Sharing the IProducer instance

Sometimes you may need to make use of already existing Confluent.Kafka.IProducer instance (i.e. for integration with existing code).

Each of the KafkaProducer methods has an overload accepting IProducer as a parameter.

Consumer

A consumer subscribes to Kafka topics and passes the messages into an Akka Stream.

Settings

When creating a consumer stream you need to pass in ConsumerSettings that define things like:

• de-serializers for the keys and values
• bootstrap servers of the Kafka cluster
• group id for the consumer, note that offsets are always committed for a given consumer group
• tuning parameters

As with producer settings, they are loaded from akka.kafka.consumer of configuration file (or custom Config instance provided).

Defining Kafka Properties Directly Inside HOCON

You can embed Kafka properties directly inside the HOCON configuration by declaring them inside the kafka-clients section:

Importing Confluent.Kafka.ConsumerConfig Directly Into ConsumerSettings

Working with ConsumerConfig is a lot more convenient than having to use the ConsumerSettings.WithProperty method because you don't have to memorize all of Kafka property names. You can import ConsumerConfig directly into Akka.Streams.Kafka.ConsumerSettings by using the convenience method ConsumerSettings.WithConsumerConfig to import all of the defined Kafka properties.

Default Consumer HOCON Settings

See reference.conf for the latest on settings.

PlainSource

To consume messages without committing them you can use KafkaConsumer.PlainSource method. This will emit consumed messages of ConsumeResult type.

Note: When using this source, you need to store consumer offset externally - it does not have support of committing offsets to Kafka.

PlainExternalSource

Special source that can use an external KafkaConsumerActor. This is useful when you have a lot of manually assigned topic-partitions and want to keep only one kafka consumer.

You can create reusable consumer actor reference like this:

CommittableSource

The KafkaConsumer.CommittableSource makes it possible to commit offset positions to Kafka.

If you need to store offsets in anything other than Kafka, PlainSource should be used instead of this API.

This is useful when "at-least once delivery" is desired, as each message will likely be delivered one time but in failure cases could be duplicated.

The recommended way to handle commits is to use the built-in Committer facilities, which provide proper batching and error handling:

The Committer facilities handle batching automatically based on your CommitterSettings. You can configure batch size, parallelism, and other parameters:

WARNING: Avoid calling CommittableOffset.Commit() or CommittableOffsetBatch.Commit() directly. Always use the Committer facilities to ensure proper batching and error handling. Direct commits can lead to reduced performance and potential data loss in failure scenarios.

When using manual partition assignment or when you need more control over the commit process:

PlainPartitionedSource

The PlainPartitionedSource is a way to track automatic partition assignment from Kafka. When a topic-partition is assigned to a consumer, this source will emit tuples with the assigned topic-partition and a corresponding source of ConsumerRecords. When a topic-partition is revoked, the corresponding source completes. As of version 1.5.39, the source automatically filters out any messages from recently revoked partitions, providing better consistency during rebalancing operations.

CommitWithMetadataSource

The CommitWithMetadataSource makes it possible to add additional metadata (in the form of a string) when an offset is committed based on the record. This can be useful (for example) to store information about which node made the commit, what time the commit was made, the timestamp of the record etc.

SourceWithOffsetContext

This source emits <see cref="ConsumeResult{TKey,TValue}"/> together with the offset position as flow context, thus makes it possible to commit offset positions to Kafka. This is useful when "at-least once delivery" is desired, as each message will likely be delivered one time but in failure cases could be duplicated. It is intended to be used with KafkaProducer.FlowWithContext and/or Committer.SinkWithOffsetContext. As of version 1.5.39, this source includes improved partition handling with automatic filtering of messages from revoked partitions.

CommittableExternalSource

Like PlainExternalSource, allows to use external KafkaConsumerActor (see documentation above).

CommittablePartitionedSource

Same as PlainPartitionedSource but with committable offset support.

AtMostOnceSource

Convenience for "at-most once delivery" semantics. The offset of each message is committed to Kafka before being emitted downstream.

CommitWithMetadataPartitionedSource

The same as PlainPartitionedSource but with offset commit with metadata support.

PlainPartitionedManualOffsetSource

The PlainPartitionedManualOffsetSource is similar to PlainPartitionedSource but allows the use of an offset store outside of Kafka, while retaining the automatic partition assignment. When a topic-partition is assigned to a consumer, the getOffsetsOnAssign function will be called to retrieve the offset, followed by a seek to the correct spot in the partition.

As of version 1.5.39, this source uses IncrementalAssign internally to prevent offset resets during partition reassignment, making it more reliable for scenarios where you're managing offsets externally - in other words: the stage now remembers any previous assignments you've made.

The onRevoke function gives the consumer a chance to store any uncommitted offsets, and do any other cleanup that is required. The source also automatically filters out any messages from recently revoked partitions to maintain consistency during rebalancing.

Transactional Producers and Consumers

Are not implemented yet. Waiting for issue https://github.com/akkadotnet/Akka.Streams.Kafka/issues/85 to be resolved.

Partition Events handling

Sometimes you may need to add custom handling for partition events, like assigning partition to consumer. To do that, you will need:

1. To write custom implementation of IPartitionEventHandler interface:

Here IRestrictedConsumer is an object providing access to some limited API of internal consumer kafka client.

2. Use WithPartitionEventsHandler of Topic / TopicPartition subscriptions, like this:

Note: Your handler callbacks will be invoked in the same thread where kafka consumer is handling all events and getting messages, so be careful when using it.

Error handling

Akka.Streams.Kafka stages utilizes stream supervision deciders to dictate what happens when a failure or exception is thrown from inside the stream stage. These deciders are basically delegate functions that returns an Akka.Streams.Supervision.Directive enumeration to tell the stage how to behave when a specific exception occured during the stream lifetime.

You can read more about stream supervision strategies in the Akka documentation

NOTE:

A decider applied to a stream using .WithAttributes(ActorAttributes.CreateSupervisionStrategy(decider)) will be used for the whole stream, any exception that happened in any of the stream stages will use the same decider to determine their fault behavior.

Producer error handling

The Akka.Streams.Kafka producers are using a default convenience error handling class called Akka.Streams.Kafka.Supervision.DefaultProducerDecider. This supervision decider uses these strategies by default:

• Any ProduceException with its IsFatal flag set will use a hard wired Directive.Stop.
• Any ProduceException that is classified as a serialization error will use a Directive.Stop. This behavior can be overriden.
• Any other ProduceException will use a Directive.Stop. This behavior can be overriden.
• Any KafkaRetriableException will use a hard wired Directive.Resume. This behavior assumes that this exception is a transient exception.
• Any KafkaException will use a Directive.Stop. This behavior can be overriden.
• Any Exception will use a Directive.Stop. This behavior can be overriden.

To create a custom decider, you will need to extend the DefaultProducerDecider:

You then register this new decider on to the stream using a stream attribute:

In this case, the decider is applied only to the PlainSink stream, it is not propagated to the stream using it.

Consumer error handling

The Akka.Streams.Kafka consumers are using a default convenience error handling class called Akka.Streams.Kafka.Supervision.DefaultConsumerDecider. This supervision decider uses these strategies by default:

• Any ConsumeException with its IsFatal flag set will use a hard wired Directive.Stop.

  As of the writing of this document, there are no fatal ConsumeException. Fatal exceptions are only thrown by producers that requires idempotence guarantee or requires transactions.

• Any ConsumeException that returns an error code of ErrorCode.UnknownTopicOrPart inside a kafka stream with auto.create.topics.enable enabled will use a Directive.Resume.

  This behavior assumes that the consumer client started before the producer was running and the broker did not have the topic or partition created yet.

• Any ConsumeException that is classified as a deserialization error will use a Directive.Stop. This behavior can be overriden.

• Any other ConsumeException will use a Directive.Resume. This behavior can be overriden.

• Any KafkaRetriableException will use a hard wired Directive.Resume. This behavior assumes that this exception is a transient exception.

• Any KafkaException will use a Directive.Resume. This behavior can be overriden.

• Any Exception will use a Directive.Stop. This behavior can be overriden.

A Directive.Resume is chosen as default because a fatal and data compromising error very rarely happened during a Consumer.Consume. The most common exceptions that are thrown during a consume are kafka configuration errors.

To create a custom decider, you will need to extend the DefaultConsumerDecider:

You then register this new decider on to the stream using a stream attribute:

Handling de/serialization exceptions

In the producer side, any serialization errors will be routed to the OnSerializationError(ProduceException<K, V> exception) callback function. The original message will be embedded inside the ProduceException.DeliveryResult.Message property if analysis were needed to determine the cause of the serialization failure. A key serialization failure will have an error code of ErrorCode.Local_KeySerialization, while a value serialization failure will have an error code of ErrorCode.Local_ValueSerialization.

In the consumer side, any deserialization errors will be routed to the OnDeserializationError(ConsumeException exception) callback function. The consumed message will be embedded inside the ConsumeException.ConsumerRecord property as a ConsumeResult<byte[], byte[]> instance. You can inspect the raw byte arrays to determine the cause of the failure. A key deserialization failure will have an error code of ErrorCode.Local_KeyDeserialization, while a value deserialization failure will have an error code of ErrorCode.Local_ValueDeserialization.

Local development

There are some helpers to simplify local development

Tests: File logging

Sometimes it is useful to have all logs written to a file in addition to console.

There is a built-in file logger, that will be added to default Akka.NET loggers if you will set AKKA_STREAMS_KAFKA_TEST_FILE_LOGGING environment variable on your local system to any value.

When set, all logs will be written to logs subfolder near to your test assembly, one file per test. Here is how log file name is generated:

Tests: Kafka container reuse

By default, tests are configured to be friendly to CI - that is, before starting tests docker Kafka images will be downloaded (if not yet exist) and containers started, and after all tests finish full cleanup will be performed (except the fact that downloaded docker images will not be removed).

While this might be useful when running tests locally, there are situations when you would like to save startup/shutdown tests time by using some pre-existing container, that will be used for all test runs and will not be stopped/started each time.

To achieve that, set AKKA_STREAMS_KAFKA_TEST_CONTAINER_REUSE environment variable on your local machine to any value. This will force using existing Kafka container, listening on port 29092 . Use docker-compose up console command in the root of project folder to get this container up and running.