From 0c4ff87453599da5ea943e5b762dec5259dd07d9 Mon Sep 17 00:00:00 2001 From: Wu Sheng Date: Sat, 31 Oct 2026 17:32:30 +0800 Subject: [PATCH] Fix the consumer race in spring-kafka-1.3.x and 2.2.x scenarios Both scenarios have the race fixed for spring-kafka-2.3.x-scenario in #841. The consumer has no committed offset and starts from the latest offset, while healthCheck reported success as soon as Tomcat started. When the entry request sent "helloWorld" before the consumer got its partition, the consumer skipped it, the request waited forever on the latch and its segment was never reported (SegmentNotFoundException for GET:/case/spring-kafka-case). Report healthy only after partitions are assigned. With auto commit disabled, the container resolves and commits the position of every assigned partition before it calls this listener, so a message sent after the health check passes is always consumed. --- .../kafka/controller/CaseController.java | 22 ++++++++++++++++++- .../kafka/controller/CaseController.java | 22 ++++++++++++++++++- 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/test/plugin/scenarios/spring-kafka-1.3.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java b/test/plugin/scenarios/spring-kafka-1.3.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java index 7c3b3201be..c75422bd70 100644 --- a/test/plugin/scenarios/spring-kafka-1.3.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java +++ b/test/plugin/scenarios/spring-kafka-1.3.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java @@ -22,8 +22,10 @@ import okhttp3.Request; import okhttp3.Response; import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; @@ -43,6 +45,7 @@ import javax.annotation.PostConstruct; import java.io.IOException; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -61,6 +64,7 @@ public class CaseController { private CountDownLatch latch = new CountDownLatch(1); private String helloWorld = "helloWorld"; + private volatile boolean partitionsAssigned = false; @PostConstruct private void setUp() { @@ -91,6 +95,19 @@ private void setUpConsumer() { Deserializer stringDeserializer = new StringDeserializer(); DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory(configs, stringDeserializer, stringDeserializer); ContainerProperties props = new ContainerProperties(topicName); + // The consumer starts from the latest offset, so the case must not send before partitions are assigned. + props.setConsumerRebalanceListener(new ConsumerRebalanceListener() { + @Override + public void onPartitionsRevoked(Collection partitions) { + } + + @Override + public void onPartitionsAssigned(Collection partitions) { + if (!partitions.isEmpty()) { + partitionsAssigned = true; + } + } + }); props.setMessageListener(new AcknowledgingMessageListener() { @Override public void onMessage(ConsumerRecord data, Acknowledgment acknowledgment) { @@ -131,7 +148,10 @@ public String springKafkaConsumerPing() { @RequestMapping("/healthCheck") @ResponseBody public String healthCheck() { - return SUCCESS; + if (partitionsAssigned) { + return SUCCESS; + } + throw new RuntimeException("kafka consumer partitions not assigned"); } } diff --git a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java index a36c34cec2..f4ce3240ed 100644 --- a/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java +++ b/test/plugin/scenarios/spring-kafka-2.2.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java @@ -22,8 +22,10 @@ import okhttp3.Request; import okhttp3.Response; import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; @@ -42,6 +44,7 @@ import javax.annotation.PostConstruct; import java.io.IOException; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -60,6 +63,7 @@ public class CaseController { private CountDownLatch latch = new CountDownLatch(1); private String helloWorld = "helloWorld"; + private volatile boolean partitionsAssigned = false; @PostConstruct private void setUp() { @@ -90,6 +94,19 @@ private void setUpConsumer() { Deserializer stringDeserializer = new StringDeserializer(); DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory(configs, stringDeserializer, stringDeserializer); ContainerProperties props = new ContainerProperties(topicName); + // The consumer starts from the latest offset, so the case must not send before partitions are assigned. + props.setConsumerRebalanceListener(new ConsumerRebalanceListener() { + @Override + public void onPartitionsRevoked(Collection partitions) { + } + + @Override + public void onPartitionsAssigned(Collection partitions) { + if (!partitions.isEmpty()) { + partitionsAssigned = true; + } + } + }); props.setMessageListener(new AcknowledgingMessageListener() { @Override public void onMessage(ConsumerRecord data, Acknowledgment acknowledgment) { @@ -130,7 +147,10 @@ public String springKafkaConsumerPing() { @RequestMapping("/healthCheck") @ResponseBody public String healthCheck() { - return SUCCESS; + if (partitionsAssigned) { + return SUCCESS; + } + throw new RuntimeException("kafka consumer partitions not assigned"); } }