Idempotency

The Idempotent Consumer from the EIP patterns is used to filter out duplicate messages: it essentially acts like a Message Filter to filter out duplicates, as reported in the Camel documentation.

From the Enterprise Integration Patterns documentation: Sometimes the same message gets delivered more than once, either because the messaging system is not certain the message has been successfully delivered yet, or because the Message Channel’s quality-of-service has been lowered to improve performance. Message receivers, on the other hand, tend to assume that each message will be delivered exactly once, and tend to cause problems when they repeat processing because of repeat messages. A receiver designed as an Idempotent Receiver handles duplicate messages and prevents them from causing problems in the receiver application.

This is a very useful feature in the integration world and it is an important new feature in the camel-kafka-connector project. Apache Camel provides multiple implementation of the Idempotent Consumer, Camel-Kafka-connector supports the in Memory and Kafka implementations.

Suppose you’re using a source connector of any kind. By using the idempotency feature you’ll be able to avoid consuming the same message multiple times.

This means, in the Kafkish language, you won’t ingest the same payload multiple times in the target Kafka topic.

In the sink scenario, we’ll stream out of a Kafka topic multiple records, transform/convert/manipulate them and send them to an external system, like a messaging broker, a storage infra, a database etc.

In the Kafka topic used as source we may have multiple repeated records with the same payload or same metadata. Based on this information we can choose to skip the same records while sending data to the external system and for doing this we can leverage the idempotency feature of ckc.

The idempotency feature can be enabled through a number of configuration options. In particular we are talking about:

The in-memory approach has been provided for short running connector workload, while the kafka one is for long running/interruptable connector.

The table is self-explaining.

A typical configuration for the kafka idempotent repository approach could be:

Some of the options can be used with their default value, in this example we’re just listing them for a Kafka idempotent repository configuration.

The Kafka idempotent repository has its own Kafka consumer and producer. They don’t use the security settings of the Kafka Connect worker, nor those of a Kafka component configured in the connector, so with only camel.idempotency.kafka.bootstrap.servers the repository can only reach an unsecured listener.

To reach a cluster secured with SSL or SASL, give the repository the Kafka client properties it needs with the camel.idempotency.kafka.client. prefix. The prefix is removed and each property is applied to both the consumer and the producer of the repository, on top of camel.idempotency.kafka.bootstrap.servers (a bootstrap.servers given with the prefix takes precedence over it). Any Kafka consumer or producer property can be passed this way, except the ones the repository always sets itself: the key and value serializers and deserializers, and enable.auto.commit, which it disables.

For a cluster that requires SSL, here with client authentication:

For a cluster that requires SASL, here SCRAM over SSL:

Rather than writing passwords and JAAS configurations in clear in the connector configuration, keep them in a secret store and reference them, as the ${file:…​} placeholders above do. Kafka Connect resolves these placeholders with a config provider, declared in the worker configuration, for example the file one:

Kafka ships other config providers as well, see Configuration Providers in the Kafka documentation. The security model describes how secrets in the connector configuration are handled.

When the connector starts, the values of all the camel.idempotency.kafka.client. properties are masked in the configuration it logs, whatever their name. The consumer and the producer of the repository log their own Kafka configuration when they are created, with the passwords and the JAAS configuration hidden by Kafka.