From 84cb7cd72e8b8d8d31a0d3438c8528173d99b8cc Mon Sep 17 00:00:00 2001 From: Carter <236375267@qq.com> Date: Fri, 30 Oct 2026 11:51:57 -0400 Subject: [PATCH 1/3] Fix NPE and lost trace context of reactive Lettuce commands when tracing is enabled Put the snapshot on the command that carries the span instead of the TracedCommand wrapper, and skip the command interceptors for a command without a span. --- CHANGES.md | 2 + .../common/RedisChannelWriterInterceptor.java | 2 +- .../RedisCommandCancelMethodInterceptor.java | 8 +- ...ompleteExceptionallyMethodInterceptor.java | 8 +- ...RedisCommandCompleteMethodInterceptor.java | 8 +- .../common/RedisCommandEnhanceInfo.java | 10 + ...disSubscriptionConstructorInterceptor.java | 6 +- .../common/RedisReactiveCommandSpanTest.java | 227 ++++++++++++++++++ 8 files changed, 256 insertions(+), 15 deletions(-) create mode 100644 apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/test/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisReactiveCommandSpanTest.java diff --git a/CHANGES.md b/CHANGES.md index d6426e015b..b5bcc6ad61 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -7,6 +7,8 @@ Release Notes. * Fix the `NullPointerException` thrown by the `postgresql-8.x` plugin when executing statements prepared with `Statement.RETURN_GENERATED_KEYS` or column names (apache/skywalking#14138). +* Fix the `NullPointerException` thrown by the Lettuce plugin and the lost trace context of reactive commands when + Lettuce tracing is enabled, and when a reactive command is cancelled before it is written (apache/skywalking#14178). All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/267?closed=1) diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptor.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptor.java index 9a94e186d8..19f9ab2a5e 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptor.java @@ -154,7 +154,7 @@ protected String getCommandName(ProtocolKeyword protocol) { return "UNKNOWN"; } - private static RedisCommand getSpanCarrierCommand(Object o) { + static RedisCommand getSpanCarrierCommand(Object o) { RedisCommand command = null; if (o instanceof RedisCommand) { command = (RedisCommand) o; diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCancelMethodInterceptor.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCancelMethodInterceptor.java index 82f4d6e55d..fd29897ca2 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCancelMethodInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCancelMethodInterceptor.java @@ -36,8 +36,8 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) { - if (objInst.getSkyWalkingDynamicField() != null) { - AbstractSpan span = ((RedisCommandEnhanceInfo) objInst.getSkyWalkingDynamicField()).getSpan(); + AbstractSpan span = RedisCommandEnhanceInfo.getSpan(objInst); + if (span != null) { span.errorOccurred(); span.tag(new StringTag(CANCEL_SIGNAL_TAG), COMMAND_CANCEL_VALUE); span.asyncFinish(); @@ -48,8 +48,8 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Throwable t) { - if (objInst.getSkyWalkingDynamicField() != null) { - AbstractSpan span = ((RedisCommandEnhanceInfo) objInst.getSkyWalkingDynamicField()).getSpan(); + AbstractSpan span = RedisCommandEnhanceInfo.getSpan(objInst); + if (span != null) { span.log(t); } } diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteExceptionallyMethodInterceptor.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteExceptionallyMethodInterceptor.java index 022c2622ee..d00c950ef0 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteExceptionallyMethodInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteExceptionallyMethodInterceptor.java @@ -33,9 +33,9 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) { - if (objInst.getSkyWalkingDynamicField() != null) { + AbstractSpan span = RedisCommandEnhanceInfo.getSpan(objInst); + if (span != null) { Throwable t = (Throwable) allArguments[0]; - AbstractSpan span = ((RedisCommandEnhanceInfo) objInst.getSkyWalkingDynamicField()).getSpan(); span.log(t); span.asyncFinish(); objInst.setSkyWalkingDynamicField(null); @@ -45,8 +45,8 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Throwable t) { - if (objInst.getSkyWalkingDynamicField() != null) { - AbstractSpan span = ((RedisCommandEnhanceInfo) objInst.getSkyWalkingDynamicField()).getSpan(); + AbstractSpan span = RedisCommandEnhanceInfo.getSpan(objInst); + if (span != null) { span.log(t); } } diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteMethodInterceptor.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteMethodInterceptor.java index 725f2742bd..9ea73190b0 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteMethodInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteMethodInterceptor.java @@ -33,8 +33,8 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) { - if (objInst.getSkyWalkingDynamicField() != null) { - AbstractSpan span = ((RedisCommandEnhanceInfo) objInst.getSkyWalkingDynamicField()).getSpan(); + AbstractSpan span = RedisCommandEnhanceInfo.getSpan(objInst); + if (span != null) { span.asyncFinish(); objInst.setSkyWalkingDynamicField(null); } @@ -44,8 +44,8 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Throwable t) { - if (objInst.getSkyWalkingDynamicField() != null) { - AbstractSpan span = ((RedisCommandEnhanceInfo) objInst.getSkyWalkingDynamicField()).getSpan(); + AbstractSpan span = RedisCommandEnhanceInfo.getSpan(objInst); + if (span != null) { span.log(t); } } diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandEnhanceInfo.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandEnhanceInfo.java index 942d5e18f4..e797d31692 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandEnhanceInfo.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandEnhanceInfo.java @@ -20,6 +20,7 @@ import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; /** * RedisCommandEnhanceInfo holds SkyWalking tracing data for Lettuce commands @@ -43,6 +44,15 @@ public AbstractSpan getSpan() { return span; } + /** + * @return the span of the command, or null if the command has no span. A reactive command only carries the + * snapshot until it is written, so a command cancelled before that has no span to finish. + */ + static AbstractSpan getSpan(EnhancedInstance command) { + RedisCommandEnhanceInfo enhanceInfo = (RedisCommandEnhanceInfo) command.getSkyWalkingDynamicField(); + return enhanceInfo == null ? null : enhanceInfo.getSpan(); + } + public RedisCommandEnhanceInfo setSpan(AbstractSpan span) { this.span = span; return this; diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisSubscriptionConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisSubscriptionConstructorInterceptor.java index 7c79ebde93..31f4d07078 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisSubscriptionConstructorInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisSubscriptionConstructorInterceptor.java @@ -25,7 +25,9 @@ * Interceptor for RedisSubscription constructor. *

* This interceptor captures the {@link io.lettuce.core.protocol.RedisCommand} instance - * at subscription construction time and stores it into SkyWalking dynamic field. + * at subscription construction time and stores it into SkyWalking dynamic field. A decorated + * command is unwrapped to the command that carries the span, so that the snapshot is found + * by {@link RedisChannelWriterInterceptor}. *

*/ public class RedisSubscriptionConstructorInterceptor implements InstanceConstructorInterceptor { @@ -33,7 +35,7 @@ public class RedisSubscriptionConstructorInterceptor implements InstanceConstruc @Override public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { // allArguments[1] is the RedisCommand passed to the RedisSubscription constructor - objInst.setSkyWalkingDynamicField(allArguments[1]); + objInst.setSkyWalkingDynamicField(RedisChannelWriterInterceptor.getSpanCarrierCommand(allArguments[1])); } } diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/test/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisReactiveCommandSpanTest.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/test/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisReactiveCommandSpanTest.java new file mode 100644 index 0000000000..857b5d8f9b --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/test/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisReactiveCommandSpanTest.java @@ -0,0 +1,227 @@ +/* + * 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.skywalking.apm.plugin.lettuce.common; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.CoreMatchers.nullValue; +import static org.hamcrest.CoreMatchers.sameInstance; +import static org.hamcrest.MatcherAssert.assertThat; + +import io.lettuce.core.protocol.Command; +import io.lettuce.core.protocol.CommandType; +import io.lettuce.core.protocol.RedisCommand; +import io.lettuce.core.protocol.TracedCommand; +import io.lettuce.core.tracing.TraceContext; +import java.util.List; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan; +import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.agent.test.helper.SegmentHelper; +import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule; +import org.apache.skywalking.apm.agent.test.tools.SegmentStorage; +import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint; +import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.reactivestreams.Subscription; +import reactor.core.CoreSubscriber; +import reactor.util.context.Context; + +@RunWith(TracingSegmentRunner.class) +public class RedisReactiveCommandSpanTest { + + private static final String PEER = "192.168.1.12:6379"; + + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + @Rule + public AgentServiceRule serviceRule = new AgentServiceRule(); + + private final RedisCommandCompleteMethodInterceptor completeInterceptor = new RedisCommandCompleteMethodInterceptor(); + private final RedisCommandCancelMethodInterceptor cancelInterceptor = new RedisCommandCancelMethodInterceptor(); + private final RedisCommandCompleteExceptionallyMethodInterceptor completeExceptionallyInterceptor = + new RedisCommandCompleteExceptionallyMethodInterceptor(); + + private MockCommand command; + private MockTracedCommand tracedCommand; + private MockInstance channelWriter; + private MockInstance subscription; + + private static class MockInstance implements EnhancedInstance { + private Object object; + + @Override + public Object getSkyWalkingDynamicField() { + return object; + } + + @Override + public void setSkyWalkingDynamicField(Object value) { + this.object = value; + } + } + + private static class MockCommand extends Command implements EnhancedInstance { + private Object object; + + MockCommand() { + super(CommandType.SET, null); + } + + @Override + public Object getSkyWalkingDynamicField() { + return object; + } + + @Override + public void setSkyWalkingDynamicField(Object value) { + this.object = value; + } + } + + private static class MockTracedCommand extends TracedCommand implements EnhancedInstance { + private Object object; + + MockTracedCommand(MockCommand command) { + super(command, TraceContext.EMPTY); + } + + @Override + public Object getSkyWalkingDynamicField() { + return object; + } + + @Override + public void setSkyWalkingDynamicField(Object value) { + this.object = value; + } + } + + private static class MockSubscriber implements CoreSubscriber { + private final ContextSnapshot snapshot; + + MockSubscriber(ContextSnapshot snapshot) { + this.snapshot = snapshot; + } + + @Override + public Context currentContext() { + return Context.of("SKYWALKING_CONTEXT_SNAPSHOT", snapshot); + } + + @Override + public void onSubscribe(Subscription s) { + } + + @Override + public void onNext(Object o) { + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onComplete() { + } + } + + @Before + public void setUp() { + command = new MockCommand(); + tracedCommand = new MockTracedCommand(command); + channelWriter = new MockInstance(); + channelWriter.setSkyWalkingDynamicField(PEER); + subscription = new MockInstance(); + } + + @Test + public void snapshotOfDecoratedCommandIsContinuedOnSpanCarrierCommand() { + subscribe(tracedCommand); + write(tracedCommand); + + assertThat(tracedCommand.getSkyWalkingDynamicField(), nullValue()); + + completeInterceptor.afterMethod(command, null, null, null, null); + // Mono#next cancels the subscription after the command has been completed. + completeInterceptor.afterMethod(tracedCommand, null, null, null, null); + cancelInterceptor.afterMethod(tracedCommand, null, null, null, null); + cancelInterceptor.afterMethod(command, null, null, null, null); + + assertThat(command.getSkyWalkingDynamicField(), nullValue()); + assertSpansOfContinuedCommand(); + } + + @Test + public void commandWithoutDecoratorStillContinuesSnapshot() { + subscribe(command); + assertThat(subscription.getSkyWalkingDynamicField(), sameInstance((Object) command)); + + write(command); + completeInterceptor.afterMethod(command, null, null, null, null); + + assertSpansOfContinuedCommand(); + } + + @Test + public void commandCancelledBeforeWriteHasNoSpanToFinish() { + subscribe(command); + Throwable error = new IllegalStateException("cancelled"); + + cancelInterceptor.afterMethod(command, null, null, null, null); + cancelInterceptor.handleMethodException(command, null, null, null, error); + completeInterceptor.afterMethod(command, null, null, null, null); + completeInterceptor.handleMethodException(command, null, null, null, error); + completeExceptionallyInterceptor.afterMethod(command, null, new Object[]{error}, null, null); + completeExceptionallyInterceptor.handleMethodException(command, null, null, null, error); + + assertThat(segmentStorage.getTraceSegments().size(), is(1)); + assertThat(SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)).size(), is(1)); + } + + private void subscribe(RedisCommand redisCommand) { + ContextManager.createEntrySpan("/redis", null); + ContextSnapshot snapshot = ContextManager.capture(); + ContextManager.stopSpan(); + + new RedisSubscriptionConstructorInterceptor().onConstruct(subscription, new Object[]{null, redisCommand}); + new RedisSubscriptionSubscribeMethodInterceptor().beforeMethod( + subscription, null, new Object[]{new MockSubscriber(snapshot)}, null, null); + } + + private void write(Object redisCommand) { + new RedisChannelWriterInterceptor().beforeMethod(channelWriter, null, new Object[]{redisCommand}, null, null); + } + + private void assertSpansOfContinuedCommand() { + List segments = segmentStorage.getTraceSegments(); + assertThat(segments.size(), is(2)); + List spans = SegmentHelper.getSpans(segments.get(1)); + assertThat(spans.size(), is(2)); + assertThat(spans.get(0).getOperationName(), is("Lettuce/UNKNOWN")); + assertThat(spans.get(0).isExit(), is(true)); + assertThat(spans.get(1).getOperationName(), is("RedisReactive/local")); + assertThat(segments.get(1).getRelatedGlobalTrace().getId(), is(segments.get(0).getRelatedGlobalTrace().getId())); + } +} From 63a9c97875ec47caa026582de521defb7b066c0d Mon Sep 17 00:00:00 2001 From: Wu Sheng Date: Sat, 31 Oct 2026 08:01:08 +0800 Subject: [PATCH 2/3] Bump skywalking-infra-e2e to fix E2E on Docker 29 runners ubuntu-latest now resolves to ubuntu-26.04, which ships Docker 29 with the containerd image store. The pinned infra-e2e action (8c21e43) disables the containerd snapshotter and restarts dockerd before running the case, so the agent image built and loaded by `make build docker.java8` (skywalking/skywalking-java:latest-java8) is no longer visible. Compose then tries to pull it from Docker Hub and both E2E cases fail with "pull access denied". Upstream infra-e2e removed that step in apache/skywalking-infra-e2e#146 and runs compose through testcontainers-go. Pin the current main (56ac8befeb5f910f2dc9339e65d07cd18953f90e), whose CI passes on ubuntu-26.04. The new action installs Go 1.26 itself. Drop the workflow's own actions/setup-go@v2 (Go 1.24) step: it exports GOROOT, which makes the action's Go 1.26 toolchain compile against the Go 1.24 standard library ("compile: version go1.24.13 does not match go tool version go1.26.10"). --- .github/workflows/e2e.yaml | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/.github/workflows/e2e.yaml b/.github/workflows/e2e.yaml index a2393b4942..23115435e2 100644 --- a/.github/workflows/e2e.yaml +++ b/.github/workflows/e2e.yaml @@ -55,12 +55,8 @@ jobs: SW_VERSION=$(./mvnw -q -DforceStdout -N org.apache.maven.plugins:maven-help-plugin:3.2.0:evaluate -Dexpression=project.version) ./mvnw -B package -f ./test/e2e/base/pom.xml -Dsw.version=${SW_VERSION} echo "::endgroup::" - - name: Setup go - uses: actions/setup-go@v2 - with: - go-version: '1.24' - name: Run E2E Tests - uses: apache/skywalking-infra-e2e@8c21e43e241a32a54bdf8eeceb9099eb27e5e9b4 + uses: apache/skywalking-infra-e2e@56ac8befeb5f910f2dc9339e65d07cd18953f90e with: e2e-file: ${{ matrix.case.path }} - uses: actions/upload-artifact@v4 From 94c442fa83a831e5c78092c4926bb400cff65e19 Mon Sep 17 00:00:00 2001 From: Wu Sheng Date: Sat, 31 Oct 2026 08:10:49 +0800 Subject: [PATCH 3/3] 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. --- .../kafka/controller/CaseController.java | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java index 0bbae644b9..7487ad5958 100644 --- a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java +++ b/test/plugin/scenarios/spring-kafka-2.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.util.Arrays; import java.io.IOException; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -62,6 +65,7 @@ public class CaseController { private CountDownLatch latch; private String helloWorld = "helloWorld"; + private volatile boolean partitionsAssigned = false; @PostConstruct private void setUp() { @@ -108,6 +112,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) { @@ -153,7 +170,10 @@ public String springKafkaConsumerPing() { @RequestMapping("/healthCheck") @ResponseBody public String healthCheck() { - return SUCCESS; + if (partitionsAssigned) { + return SUCCESS; + } + throw new RuntimeException("kafka consumer partitions not assigned"); } }