Developing Event-driven Applications with Apache Kafka and Red Hat AMQ Streams

In ACID transactional applications, the database constraints normally prevent data inconsistencies.

For example, if a user tries to create an account with the same email twice, then the database prevents that from happening due to primary or unique key constraints. The database makes the second request fail.

In event-driven applications, it is common to use an eventual consistency approach because it allows you to improve your application's performance. An eventually consistent system is a system that might show different results for the same operations, but eventually provides a consistent result.

For example, consider a system where a user can sign up from a UI. The UI sends requests to a back end with two application instances. The application saves the information in a database where data is written to the primary instance and data is read from the replicas. The primary database instance eventually updates the replicas with the latest data written. Because the update of the replicas can take some time, it is possible that the sign-up application reads inconsistent information during an undefined period of time. However, the system is eventually consistent. The following diagram shows this behavior.

Figure 6.4: Sign-up application

In an ACID database, Request 2 would have been locked until Request 1 was replicated across all replicas. Locking means that the application has to wait and, therefore, it affects the performance of the system.

Kafka Streams provides tools that might help de-duplicate input events, such as filtering transformations and tables.

Kafka consumers normally commit the offset of the messages that they consume to the Kafka broker, to keep track of what they have already processed.

By default, Kafka enables an autocommit feature for consumers. Autocommit runs every five seconds to commit the offset of the last message that the consumer has read, however, the offset is updated only when the poll method is executed. This means that although autocommit is configured to run every five seconds, if the poll method is executed every twenty seconds, then autocommit runs every five seconds committing the same offset until poll is executed again. The autocommit period is configurable.

The message delivery semantics can also be considered from the consumer side:

However, autocommitting offsets can cause problems when the application fails.

Consider a consumer using autocommit with the default period of five seconds. If the consumer takes too long processing a record and throws an exception after six seconds, although the consumer has thrown an error, the record has already been committed.

In the case of a failure, you might lose records or read duplicate records.

Depending on the business logic of your application, there are several ways in which you can achieve an exactly once message delivery semantic to avoid data duplicate and data loss.

In the consumer side, autocommit might cause both data duplication and loss. To avoid it, you can disable the autocommit feature by setting the enable.auto.commit property to false. If you want to keep it, then you can control how often it is triggered by setting the auto.commit.interval.ms property.

If you disable autocommitting, then you must manually send the offset of the consumer by using the commitSync and commitAsync methods in the KafkaConsumer class.

The commitSync method blocks the current thread until the Kafka broker returns a response. This results in higher latency because your application waits for a success or an error response, and it retries sending the commit in the case of a transient error.

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100); 
    for (ConsumerRecord<String, String> record : records) {
        // Record processing
    }

    consumer.commitSync(); 
}

The commitAsync method does not wait for the response from the broker. The consumer keeps on polling and processing records, and when the response from the broker is ready, a callback is executed to handle the response. This results in lower latency because the thread polling the records is not blocked. However, a lower offset than the real one can be committed because no order is guaranteed.

For example, if you send an asynchronous commit for offset 10 and it gets delayed because of a networking issue, then another commit for offset 15 can arrive before at the broker. The last commit request marks the offset, so offset 10 is committed although a record with offset 15 was read. This behavior causes duplicates.

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        // Record processing
    }

    consumer.commitAsync(callback);
}

If you are using Quarkus to consume Kafka messages, then you can also select the commit strategy that an incoming channel uses by setting the mp.messaging.incoming.my-channel.commit-strategy property. By default, the framework disables autocommit. There are three possible values:

If your application consumes messages from a topic, processes them, and produces the result to another topic, then you can use the transactional consume-produce feature that Kafka offers. A transaction makes the consuming, processing, and producing steps atomic, which means that either all records are consumed, processed, and produced or none are.

For example, consider that your application consumes records from the A topic, processes the records, and produces those records to the B topic. If your application has consumed, processed, and produced record 1, but it fails while processing record 2, then record 1 is recovered.

The producer does most of the work in the transaction. You should assign a transactional identifier to producers, this is used to uniquely identify a producer. If a producer crashes and leaves uncompleted transactions, then the transactional identifier is used to identify what producer started the transaction.

The following snippet shows a transaction implementation.

KafkaProducer producer = createProducer(“transactional.id”, “my-transactional-id”); 

producer.initTransactions(); 

KafkaConsumer consumer = createConsumer(
  “bootstrap.servers”, “localhost:9092”,
  “group.id”, “my-group-id”); 

consumer.subscribe(singleton(“inputTopic”));

while (true) {
  ConsumerRecords records = consumer.poll(Long.MAX_VALUE);

  producer.beginTransaction(); 
  for (ConsumerRecord record : records) {
    // Processing
    producer.send(producerRecord(“outputTopic”, record)); 
  }
  producer.sendOffsetsToTransaction(currentOffsets(consumer), group); 
  producer.commitTransaction(); 
}

If something goes wrong between the begin and commit stages, the transaction is aborted. The application leaves the offset commit log and the topics intact. This protects the application against duplicate messages published to the output topic, and lost messages from the input topic.

Because of the consume, process, produce nature of Kafka transactions, Kafka Streams uses them internally.

Kafka Streams supports exactly-once processing (EOS). To achieve this message delivery semantic, Kafka Streams uses the idempotence and transaction features provided by Kafka.

You can enable exactly-once processing by setting the processing.guarantee configuration value to exactly_once or exactly_once_beta. The exactly_once_beta value improves the performance and requires a Kafka version higher than 2.5. The default value for this property is at_least_once.

Enabling the exactly once guarantee changes the configuration of some Kafka configuration values, such as enable.idempotence and commit.interval.ms.