Skip to content

Commit 94c442f

Browse files
committed
Fix the consumer race in spring-kafka-2.3.x-scenario
The scenario consumer has no committed offset and starts from the latest offset, while healthCheck reported success as soon as Tomcat started. When the health check hit right after startup, the entry request sent "helloWorld" before the consumer got its partition, so the consumer skipped it, the request waited forever on the latch and its segment was never reported (SegmentNotFoundException for GET:/case/spring-kafka-case). On ubuntu-26.04 runners spring-kafka 2.7.14 lost this race on every attempt. Report healthy only after partitions are assigned, the same way kafka-scenario does.
1 parent 63a9c97 commit 94c442f

1 file changed

Lines changed: 21 additions & 1 deletion

File tree

  • test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller

‎test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,10 @@
2222
import okhttp3.Request;
2323
import okhttp3.Response;
2424
import org.apache.kafka.clients.consumer.ConsumerConfig;
25+
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
2526
import org.apache.kafka.clients.consumer.ConsumerRecord;
2627
import org.apache.kafka.clients.producer.ProducerConfig;
28+
import org.apache.kafka.common.TopicPartition;
2729
import org.apache.kafka.common.serialization.Deserializer;
2830
import org.apache.kafka.common.serialization.StringDeserializer;
2931
import org.apache.kafka.common.serialization.StringSerializer;
@@ -43,6 +45,7 @@
4345
import javax.annotation.PostConstruct;
4446
import java.util.Arrays;
4547
import java.io.IOException;
48+
import java.util.Collection;
4649
import java.util.HashMap;
4750
import java.util.Map;
4851
import java.util.concurrent.CountDownLatch;
@@ -62,6 +65,7 @@ public class CaseController {
6265

6366
private CountDownLatch latch;
6467
private String helloWorld = "helloWorld";
68+
private volatile boolean partitionsAssigned = false;
6569

6670
@PostConstruct
6771
private void setUp() {
@@ -108,6 +112,19 @@ private void setUpConsumer() {
108112
Deserializer<String> stringDeserializer = new StringDeserializer();
109113
DefaultKafkaConsumerFactory<String, String> factory = new DefaultKafkaConsumerFactory(configs, stringDeserializer, stringDeserializer);
110114
ContainerProperties props = new ContainerProperties(topicName);
115+
// The consumer starts from the latest offset, so the case must not send before partitions are assigned.
116+
props.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
117+
@Override
118+
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
119+
}
120+
121+
@Override
122+
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
123+
if (!partitions.isEmpty()) {
124+
partitionsAssigned = true;
125+
}
126+
}
127+
});
111128
props.setMessageListener(new AcknowledgingMessageListener<String, String>() {
112129
@Override
113130
public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment) {
@@ -153,7 +170,10 @@ public String springKafkaConsumerPing() {
153170
@RequestMapping("/healthCheck")
154171
@ResponseBody
155172
public String healthCheck() {
156-
return SUCCESS;
173+
if (partitionsAssigned) {
174+
return SUCCESS;
175+
}
176+
throw new RuntimeException("kafka consumer partitions not assigned");
157177
}
158178
}
159179

0 commit comments

Comments
 (0)