CAMEL-16914: simplify handling of numeric headers for idempotency support

components/camel-kafka/src/main/docs/kafka-component.adoc

@@ -530,6 +530,7 @@ The `camel-kafka` library provides a Kafka topic-based idempotent repository. Th
 The topic used must be unique per idempotent repository instance. The mechanism does not have any requirements about the number of topic partitions; as the repository consumes from all partitions at the same time. It also does not have any requirements about the replication factor of the topic.
 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"]
@@ -593,6 +594,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
 
 By default the Kafka consumer will use auto commit, where the offset will be committed automatically in the background using a given interval.

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());
+            }
+        });
+    }
+}

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);
+        }
+    }
+}

components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java

@@ -37,12 +37,15 @@
 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";

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);
+    }
+}

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");
+    }
+}