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.

Author: wu-sheng

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<String> stringDeserializer = new StringDeserializer();
         DefaultKafkaConsumerFactory<String, String> 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<TopicPartition> partitions) {
+            }
+
+            @Override
+            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+                if (!partitions.isEmpty()) {
+                    partitionsAssigned = true;
+                }
+            }
+        });
         props.setMessageListener(new AcknowledgingMessageListener<String, String>() {
             @Override
             public void onMessage(ConsumerRecord<String, String> 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");
     }
 }
 

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<String> stringDeserializer = new StringDeserializer();
         DefaultKafkaConsumerFactory<String, String> 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<TopicPartition> partitions) {
+            }
+
+            @Override
+            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+                if (!partitions.isEmpty()) {
+                    partitionsAssigned = true;
+                }
+            }
+        });
         props.setMessageListener(new AcknowledgingMessageListener<String, String>() {
             @Override
             public void onMessage(ConsumerRecord<String, String> 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");
     }
 }