[camel] branch main updated: Test manual commit when rebalancing in kafka occur (#5630)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new 1313d51  Test manual commit when rebalancing in kafka occur (#5630)
1313d51 is described below

commit 1313d51259c8f5d3c4768bd9c81282607365bb1a
Author: Federico Mariani <34...@users.noreply.github.com>
AuthorDate: Tue Jun 8 17:32:48 2021 +0200

    Test manual commit when rebalancing in kafka occur (#5630)
    
    Co-authored-by: Federico Mariani <fm...@redhat.com>
---
 .../integration/KafkaConsumerManualCommitIT.java   | 52 ++++++++++++++++++++++
 1 file changed, 52 insertions(+)

diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerManualCommitIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerManualCommitIT.java
index 5fd9de8..c3c45f6 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerManualCommitIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerManualCommitIT.java
@@ -44,6 +44,9 @@ public class KafkaConsumerManualCommitIT extends BaseEmbeddedKafkaTestSupport {
     @EndpointInject("mock:result")
     private MockEndpoint to;
 
+    @EndpointInject("mock:resultBar")
+    private MockEndpoint toBar;
+
     private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
 
     @BeforeEach
@@ -72,11 +75,60 @@ public class KafkaConsumerManualCommitIT extends BaseEmbeddedKafkaTestSupport {
                     assertNotNull(manual);
                     manual.commitSync();
                 });
+                from(from).routeId("bar").autoStartup(false).to(toBar);
             }
         };
     }
 
     @Test
+    public void kafkaAutoCommitDisabledDuringRebalance() throws Exception {
+        to.expectedMessageCount(1);
+        String firstMessage = "message-0";
+        to.expectedBodiesReceivedInAnyOrder(firstMessage);
+
+        ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC, "1", firstMessage);
+        producer.send(data);
+
+        to.assertIsSatisfied(3000);
+
+        to.reset();
+
+        context.getRouteController().stopRoute("foo");
+        to.expectedMessageCount(0);
+
+        String secondMessage = "message-1";
+        data = new ProducerRecord<>(TOPIC, "1", secondMessage);
+        producer.send(data);
+
+        to.assertIsSatisfied(3000);
+
+        to.reset();
+
+        // start a new route in order to rebalance kafka
+        context.getRouteController().startRoute("bar");
+        toBar.expectedMessageCount(1);
+        synchronized (this) {
+            Thread.sleep(1000);
+        }
+
+        toBar.assertIsSatisfied();
+
+        context.getRouteController().stopRoute("bar");
+
+        // The route bar is not committing the offset, so by restarting foo, last 3 items will be processed
+        context.getRouteController().startRoute("foo");
+        to.expectedMessageCount(1);
+        to.expectedBodiesReceivedInAnyOrder("message-1");
+
+        // give some time for the route to start again
+        synchronized (this) {
+            Thread.sleep(1000);
+        }
+
+        to.assertIsSatisfied(3000);
+    }
+
+    @Test
     public void kafkaManualCommit() throws Exception {
         to.expectedMessageCount(5);
         to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");