This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-3.11.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-3.11.x by this push:
new 8b3d6e2 CAMEL-16914: simplify handling of numeric headers for
idempotency support (#6053) (#6062)
8b3d6e2 is described below
commit 8b3d6e290e74661f22d442f16a12bf5ba79367bb
Author: Otavio Rodolfo Piske <[email protected]>
AuthorDate: Thu Sep 9 12:54:23 2021 +0200
CAMEL-16914: simplify handling of numeric headers for idempotency support
(#6053) (#6062)
---
.../camel-kafka/src/main/docs/kafka-component.adoc | 47 +++++++++++
.../component/kafka/serde/KafkaSerdeHelper.java | 43 ++++++++++
.../integration/CustomHeaderDeserializer.java | 40 +++++++++
.../kafka/integration/KafkaConsumerFullIT.java | 3 +
.../integration/KafkaConsumerIdempotentIT.java | 91 +++++++++++++++++++++
.../KafkaConsumerIdempotentTestSupport.java | 69 ++++++++++++++++
...kaConsumerIdempotentWithCustomSerializerIT.java | 87 ++++++++++++++++++++
.../KafkaConsumerIdempotentWithProcessorIT.java | 94 ++++++++++++++++++++++
.../integration/KafkaConsumerTopicIsPatternIT.java | 3 +
.../modules/ROOT/pages/kafka-component.adoc | 47 +++++++++++
10 files changed, 524 insertions(+)
diff --git a/components/camel-kafka/src/main/docs/kafka-component.adoc
b/components/camel-kafka/src/main/docs/kafka-component.adoc
index 940e9b7..f42c987 100644
--- a/components/camel-kafka/src/main/docs/kafka-component.adoc
+++ b/components/camel-kafka/src/main/docs/kafka-component.adoc
@@ -488,6 +488,7 @@ The topic used must be unique per idempotent repository
instance. The mechanism
Each repository instance that uses the topic (e.g. typically on different
machines running in parallel) controls its own consumer group, so in a cluster
of 10 Camel processes using the same topic each will control its own offset.
On startup, the instance subscribes to the topic and rewinds the offset to the
beginning, rebuilding the cache to the latest state. The cache will not be
considered warmed up until one poll of `pollDurationMs` in length returns 0
records. Startup will not be completed until either the cache has warmed up, or
30 seconds go by; if the latter happens the idempotent repository may be in an
inconsistent state until its consumer catches up to the end of the topic.
+Be mindful of the format of the header used for the uniqueness check. By
default, it uses Strings as the data types. When using primitive numeric
formats, the header must be deserialized accordingly. Check the samples below
for examples.
A `KafkaIdempotentRepository` has the following properties:
[width="100%",cols="2m,5",options="header"]
@@ -552,6 +553,52 @@ In XML:
</bean>
----
+There are 3 alternatives to choose from when using idempotency with numeric
identifiers. The first one is to use the static method `numericHeader` method
from `org.apache.camel.component.kafka.serde.KafkaSerdeHelper` to perform the
conversion for you:
+
+[source,java]
+----
+from("direct:performInsert")
+
.idempotentConsumer(numericHeader("id")).messageIdRepositoryRef("insertDbIdemRepo")
+ // once-only insert into database
+ .end()
+----
+
+Alternatively, it is possible use a custom serializer configured via the route
URL to perform the conversion:
+
+[source,java]
+----
+public class CustomHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
+ private static final Logger LOG =
LoggerFactory.getLogger(CustomHeaderDeserializer.class);
+
+ @Override
+ public Object deserialize(String key, byte[] value) {
+ if (key.equals("id")) {
+ BigInteger bi = new BigInteger(value);
+
+ return String.valueOf(bi.longValue());
+ } else {
+ return super.deserialize(key, value);
+ }
+ }
+}
+----
+
+Lastly, it is also possible to do so in a processor:
+
+[source,java]
+----
+from(from).routeId("foo")
+ .process(exchange -> {
+ byte[] id = exchange.getIn().getHeader("id", byte[].class);
+
+ BigInteger bi = new BigInteger(id);
+ exchange.getIn().setHeader("id", String.valueOf(bi.longValue()));
+ })
+ .idempotentConsumer(header("id"))
+ .messageIdRepositoryRef("kafkaIdempotentRepository")
+ .to(to);
+----
+
== Using manual commit with Kafka consumer
*Since Camel 2.21*
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/serde/KafkaSerdeHelper.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/serde/KafkaSerdeHelper.java
new file mode 100644
index 0000000..645da01
--- /dev/null
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/serde/KafkaSerdeHelper.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.component.kafka.serde;
+
+import java.math.BigInteger;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.support.ExpressionAdapter;
+import org.apache.camel.support.builder.ValueBuilder;
+
+public final class KafkaSerdeHelper {
+ private KafkaSerdeHelper() {
+
+ }
+
+ public static ValueBuilder numericHeader(String name) {
+ return new ValueBuilder(new ExpressionAdapter() {
+ @Override
+ public Object evaluate(Exchange exchange) {
+ byte[] id = exchange.getIn().getHeader(name, byte[].class);
+
+ BigInteger bi = new BigInteger(id);
+
+ return String.valueOf(bi.longValue());
+ }
+ });
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/CustomHeaderDeserializer.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/CustomHeaderDeserializer.java
new file mode 100644
index 0000000..44c295a
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/CustomHeaderDeserializer.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.component.kafka.integration;
+
+import java.math.BigInteger;
+
+import org.apache.camel.component.kafka.serde.DefaultKafkaHeaderDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CustomHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
+ private static final Logger LOG =
LoggerFactory.getLogger(CustomHeaderDeserializer.class);
+
+ @Override
+ public Object deserialize(String key, byte[] value) {
+ if (key.equals("id")) {
+ BigInteger bi = new BigInteger(value);
+ LOG.debug("Converted the header {} to {} via custom serializer",
key, bi.longValue());
+
+ return String.valueOf(bi.longValue());
+ } else {
+ return super.deserialize(key, value);
+ }
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
index 3940b6e..5cdcc35 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
@@ -37,12 +37,15 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
import static org.apache.camel.test.junit5.TestSupport.assertIsInstanceOf;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+@DisabledIfSystemProperty(named = "enable.kafka.consumer.idempotency.tests",
matches = "true",
+ disabledReason = "Runtime conflicts with the
idempotency tests")
public class KafkaConsumerFullIT extends BaseEmbeddedKafkaTestSupport {
public static final String TOPIC = "test";
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentIT.java
new file mode 100644
index 0000000..058b885
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentIT.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.integration;
+
+import java.util.Collections;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Endpoint;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+
+import static
org.apache.camel.component.kafka.serde.KafkaSerdeHelper.numericHeader;
+
+@EnabledIfSystemProperty(named = "enable.kafka.consumer.idempotency.tests",
matches = "true")
+public class KafkaConsumerIdempotentIT extends
KafkaConsumerIdempotentTestSupport {
+
+ public static final String TOPIC = "idempt";
+
+ @BindToRegistry("kafkaIdempotentRepository")
+ private KafkaIdempotentRepository kafkaIdempotentRepository
+ = new KafkaIdempotentRepository("TEST_IDEMPOTENT",
getBootstrapServers());
+
+ @EndpointInject("kafka:" + TOPIC
+ + "?groupId=group2&autoOffsetReset=earliest"
+ +
"&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ +
"&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ +
"&autoCommitIntervalMs=1000&sessionTimeoutMs=30000&autoCommitEnable=true"
+ +
"&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
+ private Endpoint from;
+
+ @EndpointInject("mock:result")
+ private MockEndpoint to;
+
+ private int size = 200;
+
+ @BeforeEach
+ public void before() throws ExecutionException, InterruptedException,
TimeoutException {
+ doSend(size, TOPIC);
+ }
+
+ @AfterEach
+ public void after() {
+
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() throws Exception {
+ from(from).routeId("foo")
+ .idempotentConsumer(numericHeader("id"))
+ .messageIdRepositoryRef("kafkaIdempotentRepository")
+ .to(to);
+ }
+ };
+ }
+
+ @Test
+ @DisplayName("Numeric headers is consumable when using idempotent
(CAMEL-16914)")
+ public void kafkaIdempotentMessageIsConsumedByCamel() throws
InterruptedException {
+ doRun(to, size);
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentTestSupport.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentTestSupport.java
new file mode 100644
index 0000000..e3f62d6
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentTestSupport.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.component.kafka.integration;
+
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.internals.RecordHeader;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public abstract class KafkaConsumerIdempotentTestSupport extends
BaseEmbeddedKafkaTestSupport {
+
+ protected void doSend(int size, String topic) throws ExecutionException,
InterruptedException, TimeoutException {
+ Properties props = getDefaultProperties();
+ org.apache.kafka.clients.producer.KafkaProducer<String, String>
producer
+ = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
+
+ try {
+ for (int k = 0; k < size; k++) {
+ String msg = "message-" + k;
+ ProducerRecord<String, String> data = new
ProducerRecord<>(topic, String.valueOf(k), msg);
+
+ data.headers().add(new RecordHeader("id",
BigInteger.valueOf(k).toByteArray()));
+ producer.send(data);
+ }
+ } finally {
+ if (producer != null) {
+ producer.close();
+ }
+ }
+ }
+
+ protected void doRun(MockEndpoint mockEndpoint, int size) throws
InterruptedException {
+ mockEndpoint.expectedMessageCount(size);
+
+ List<Exchange> exchangeList = mockEndpoint.getReceivedExchanges();
+
+ mockEndpoint.assertIsSatisfied(10000);
+
+ assertEquals(size, exchangeList.size());
+
+ Map<String, Object> headers =
mockEndpoint.getExchanges().get(0).getIn().getHeaders();
+ assertTrue(headers.containsKey("id"), "0");
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithCustomSerializerIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithCustomSerializerIT.java
new file mode 100644
index 0000000..5a37f0d
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithCustomSerializerIT.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.integration;
+
+import java.util.Collections;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Endpoint;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+
+@EnabledIfSystemProperty(named = "enable.kafka.consumer.idempotency.tests",
matches = "true")
+public class KafkaConsumerIdempotentWithCustomSerializerIT extends
KafkaConsumerIdempotentTestSupport {
+
+ public static final String TOPIC = "idempt2";
+
+ @BindToRegistry("kafkaIdempotentRepository")
+ private KafkaIdempotentRepository kafkaIdempotentRepository
+ = new KafkaIdempotentRepository("TEST_IDEMPOTENT",
getBootstrapServers());
+
+ @EndpointInject("kafka:" + TOPIC
+ + "?groupId=group2&autoOffsetReset=earliest"
+ +
"&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ +
"&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ +
"&headerDeserializer=#class:org.apache.camel.component.kafka.integration.CustomHeaderDeserializer"
+ +
"&autoCommitIntervalMs=1000&sessionTimeoutMs=30000&autoCommitEnable=true"
+ +
"&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
+ private Endpoint from;
+
+ @EndpointInject("mock:result")
+ private MockEndpoint to;
+
+ private int size = 200;
+
+ @BeforeEach
+ public void before() throws ExecutionException, InterruptedException,
TimeoutException {
+ doSend(size, TOPIC);
+ }
+
+ @AfterEach
+ public void after() {
+
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(from).routeId("foo")
+ .idempotentConsumer(header("id"))
+ .messageIdRepositoryRef("kafkaIdempotentRepository")
+ .to(to);
+ }
+ };
+ }
+
+ @Test
+ public void kafkaMessageIsConsumedByCamel() throws InterruptedException {
+ doRun(to, size);
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithProcessorIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithProcessorIT.java
new file mode 100644
index 0000000..4f0bd9e
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerIdempotentWithProcessorIT.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.integration;
+
+import java.math.BigInteger;
+import java.util.Collections;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Endpoint;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+
+@EnabledIfSystemProperty(named = "enable.kafka.consumer.idempotency.tests",
matches = "true")
+public class KafkaConsumerIdempotentWithProcessorIT extends
KafkaConsumerIdempotentTestSupport {
+ public static final String TOPIC = "testidemp3";
+
+ @BindToRegistry("kafkaIdempotentRepository")
+ private KafkaIdempotentRepository kafkaIdempotentRepository
+ = new KafkaIdempotentRepository("TEST_IDEMPOTENT",
getBootstrapServers());
+
+ @EndpointInject("kafka:" + TOPIC
+ + "?groupId=group2&autoOffsetReset=earliest"
+ +
"&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ +
"&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ +
"&autoCommitIntervalMs=1000&sessionTimeoutMs=30000&autoCommitEnable=true"
+ +
"&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
+ private Endpoint from;
+
+ @EndpointInject("mock:result")
+ private MockEndpoint to;
+
+ private int size = 200;
+
+ @BeforeEach
+ public void before() throws ExecutionException, InterruptedException,
TimeoutException {
+ doSend(size, TOPIC);
+ }
+
+ @AfterEach
+ public void after() {
+
+ // clean all test topics
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+
+ @Override
+ public void configure() throws Exception {
+ from(from).routeId("foo")
+ .process(exchange -> {
+ byte[] id = exchange.getIn().getHeader("id",
byte[].class);
+
+ BigInteger bi = new BigInteger(id);
+
+ exchange.getIn().setHeader("id",
String.valueOf(bi.longValue()));
+ })
+ .idempotentConsumer(header("id"))
+ .messageIdRepositoryRef("kafkaIdempotentRepository")
+ .to(to);
+ }
+ };
+ }
+
+ @Test
+ public void kafkaMessageIsConsumedByCamel() throws InterruptedException {
+ doRun(to, size);
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
index 50281ed..af9c8ce 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerTopicIsPatternIT.java
@@ -30,9 +30,12 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
import static org.junit.jupiter.api.Assertions.assertEquals;
+@DisabledIfSystemProperty(named = "enable.kafka.consumer.idempotency.tests",
matches = "true",
+ disabledReason = "Runtime conflicts with the
idempotency tests")
public class KafkaConsumerTopicIsPatternIT extends
BaseEmbeddedKafkaTestSupport {
public static final String TOPIC = "test";
diff --git a/docs/components/modules/ROOT/pages/kafka-component.adoc
b/docs/components/modules/ROOT/pages/kafka-component.adoc
index 442f61a..bc95414 100644
--- a/docs/components/modules/ROOT/pages/kafka-component.adoc
+++ b/docs/components/modules/ROOT/pages/kafka-component.adoc
@@ -490,6 +490,7 @@ The topic used must be unique per idempotent repository
instance. The mechanism
Each repository instance that uses the topic (e.g. typically on different
machines running in parallel) controls its own consumer group, so in a cluster
of 10 Camel processes using the same topic each will control its own offset.
On startup, the instance subscribes to the topic and rewinds the offset to the
beginning, rebuilding the cache to the latest state. The cache will not be
considered warmed up until one poll of `pollDurationMs` in length returns 0
records. Startup will not be completed until either the cache has warmed up, or
30 seconds go by; if the latter happens the idempotent repository may be in an
inconsistent state until its consumer catches up to the end of the topic.
+Be mindful of the format of the header used for the uniqueness check. By
default, it uses Strings as the data types. When using primitive numeric
formats, the header must be deserialized accordingly. Check the samples below
for examples.
A `KafkaIdempotentRepository` has the following properties:
[width="100%",cols="2m,5",options="header"]
@@ -554,6 +555,52 @@ In XML:
</bean>
----
+There are 3 alternatives to choose from when using idempotency with numeric
identifiers. The first one is to use the static method `numericHeader` method
from `org.apache.camel.component.kafka.serde.KafkaSerdeHelper` to perform the
conversion for you:
+
+[source,java]
+----
+from("direct:performInsert")
+
.idempotentConsumer(numericHeader("id")).messageIdRepositoryRef("insertDbIdemRepo")
+ // once-only insert into database
+ .end()
+----
+
+Alternatively, it is possible use a custom serializer configured via the route
URL to perform the conversion:
+
+[source,java]
+----
+public class CustomHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
+ private static final Logger LOG =
LoggerFactory.getLogger(CustomHeaderDeserializer.class);
+
+ @Override
+ public Object deserialize(String key, byte[] value) {
+ if (key.equals("id")) {
+ BigInteger bi = new BigInteger(value);
+
+ return String.valueOf(bi.longValue());
+ } else {
+ return super.deserialize(key, value);
+ }
+ }
+}
+----
+
+Lastly, it is also possible to do so in a processor:
+
+[source,java]
+----
+from(from).routeId("foo")
+ .process(exchange -> {
+ byte[] id = exchange.getIn().getHeader("id", byte[].class);
+
+ BigInteger bi = new BigInteger(id);
+ exchange.getIn().setHeader("id", String.valueOf(bi.longValue()));
+ })
+ .idempotentConsumer(header("id"))
+ .messageIdRepositoryRef("kafkaIdempotentRepository")
+ .to(to);
+----
+
== Using manual commit with Kafka consumer
*Since Camel 2.21*
