From c65bcf1142d8c19448567ee654de17aa3014cc74 Mon Sep 17 00:00:00 2001 From: Lu Jiajing Date: Fri, 18 Sep 2020 15:18:54 +0800 Subject: [PATCH] Check bootstrapServers type before casting for kafka-clients >= 2.1 (#5510) --- .../ProducerConstructorMapInterceptor.java | 10 ++- ...ProducerConstructorMapInterceptorTest.java | 90 +++++++++++++++++++ .../config/expectedData.yaml | 74 +++++++++++++++ .../kafka/controller/CaseController.java | 27 +++++- 4 files changed, 198 insertions(+), 3 deletions(-) create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/ProducerConstructorMapInterceptorTest.java diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/ProducerConstructorMapInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/ProducerConstructorMapInterceptor.java index a6925f3ae..cd6e507ab 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/ProducerConstructorMapInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/ProducerConstructorMapInterceptor.java @@ -22,16 +22,24 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedI import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor; import org.apache.skywalking.apm.util.StringUtil; +import java.util.List; import java.util.Map; +import java.util.regex.Pattern; public class ProducerConstructorMapInterceptor implements InstanceConstructorInterceptor { + private static final Pattern COMMA_WITH_WHITESPACE = Pattern.compile("\\s*,\\s*"); @Override public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { Map config = (Map) allArguments[0]; // prevent errors caused by secondary interception in kafkaTemplate if (objInst.getSkyWalkingDynamicField() == null) { - objInst.setSkyWalkingDynamicField(StringUtil.join(';', ((String) config.get("bootstrap.servers")).split(","))); + Object bootstrapServers = config.get("bootstrap.servers"); + if (bootstrapServers instanceof List) { + objInst.setSkyWalkingDynamicField(String.join(";", (List) bootstrapServers)); + } else { + objInst.setSkyWalkingDynamicField(StringUtil.join(';', COMMA_WITH_WHITESPACE.split((String) bootstrapServers, -1))); + } } } } \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/ProducerConstructorMapInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/ProducerConstructorMapInterceptorTest.java new file mode 100644 index 000000000..8cc91452a --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/ProducerConstructorMapInterceptorTest.java @@ -0,0 +1,90 @@ +/* + * 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.kafka; + +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.core.Is.is; + +@RunWith(MockitoJUnitRunner.class) +public class ProducerConstructorMapInterceptorTest { + private static Map PRODUCER_CONFIG_WITH_LIST_BOOTSTRAP_SERVERS; + + private static Map PRODUCER_CONFIG_WITH_STRING_BOOTSTRAP_SERVERS; + + @Mock + private ProducerConstructorMapInterceptor constructorMapInterceptor; + + private EnhancedInstance enhancedInstance = new EnhancedInstance() { + private String brokerServers; + + @Override + public Object getSkyWalkingDynamicField() { + return brokerServers; + } + + @Override + public void setSkyWalkingDynamicField(Object value) { + brokerServers = (String) value; + } + }; + + @BeforeClass + public static void setup() { + List mockBootstrapServers = new ArrayList(); + mockBootstrapServers.add("localhost:9092"); + mockBootstrapServers.add("localhost:19092"); + PRODUCER_CONFIG_WITH_LIST_BOOTSTRAP_SERVERS = new HashMap() {{ + put("bootstrap.servers", mockBootstrapServers); + }}; + PRODUCER_CONFIG_WITH_STRING_BOOTSTRAP_SERVERS = new HashMap() {{ + // deliberately add whitespaces + put("bootstrap.servers", String.join(" , ", mockBootstrapServers)); + }}; + } + + @Before + public void setUpForEach() { + constructorMapInterceptor = new ProducerConstructorMapInterceptor(); + } + + @Test + public void givenListTypeBootstrapServers_whenConstructProducer_thenServersSaves() { + constructorMapInterceptor.onConstruct(enhancedInstance, new Object[]{PRODUCER_CONFIG_WITH_LIST_BOOTSTRAP_SERVERS}); + assertThat(enhancedInstance.getSkyWalkingDynamicField().toString(), is("localhost:9092;localhost:19092")); + } + + @Test + public void givenStringTypeBootstrapServers_whenConstructProducer_thenServersSaves() { + constructorMapInterceptor.onConstruct(enhancedInstance, new Object[]{PRODUCER_CONFIG_WITH_STRING_BOOTSTRAP_SERVERS}); + assertThat(enhancedInstance.getSkyWalkingDynamicField().toString(), is("localhost:9092;localhost:19092")); + } +} \ No newline at end of file diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/config/expectedData.yaml b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/config/expectedData.yaml index 74116c0df..06c2934d2 100644 --- a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/config/expectedData.yaml +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/config/expectedData.yaml @@ -34,6 +34,21 @@ segmentItems: tags: - {key: mq.broker, value: 'kafka-server:9092'} - {key: mq.topic, value: spring_test} + - operationName: Kafka/spring_test/Producer + operationId: 0 + parentSpanId: 0 + spanId: 2 + spanLayer: MQ + startTime: not null + endTime: not null + componentId: 40 + isError: false + spanType: Exit + peer: kafka-server:9092 + skipAnalysis: false + tags: + - { key: mq.broker, value: 'kafka-server:9092' } + - { key: mq.topic, value: spring_test } - operationName: /case/spring-kafka-case operationId: 0 parentSpanId: -1 @@ -71,6 +86,28 @@ segmentItems: refType: CrossProcess, parentSpanId: 1, parentTraceSegmentId: not null, parentServiceInstance: not null, parentService: spring-kafka-2.3.x-scenario, traceId: not null} + - segmentId: not null + spans: + - operationName: /case/spring-kafka-consumer-ping + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: Http + startTime: not null + endTime: not null + componentId: 14 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + tags: + - {key: url, value: 'http://localhost:8080/spring-kafka-2.3.x-scenario/case/spring-kafka-consumer-ping'} + - {key: http.method, value: GET} + refs: + - {parentEndpoint: 'Kafka/spring_test/Consumer/grop:spring_test', networkAddress: 'localhost:8080', + refType: CrossProcess, parentSpanId: 1, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: spring-kafka-2.3.x-scenario, + traceId: not null} - segmentId: not null spans: - operationName: /spring-kafka-2.3.x-scenario/case/spring-kafka-consumer-ping @@ -108,3 +145,40 @@ segmentItems: refType: CrossProcess, parentSpanId: not null, parentTraceSegmentId: not null, parentServiceInstance: not null, parentService: spring-kafka-2.3.x-scenario, traceId: not null} + - segmentId: not null + spans: + - operationName: /spring-kafka-2.3.x-scenario/case/spring-kafka-consumer-ping + operationId: 0 + parentSpanId: 0 + spanId: 1 + spanLayer: Http + startTime: not null + endTime: not null + componentId: 12 + isError: false + spanType: Exit + peer: localhost:8080 + skipAnalysis: false + tags: + - {key: http.method, value: GET} + - {key: url, value: 'http://localhost:8080/spring-kafka-2.3.x-scenario/case/spring-kafka-consumer-ping'} + - operationName: Kafka/spring_test/Consumer/grop:spring_test + operationId: 0 + parentSpanId: -1 + spanId: 0 + spanLayer: MQ + startTime: not null + endTime: not null + componentId: 41 + isError: false + spanType: Entry + peer: '' + skipAnalysis: false + tags: + - {key: mq.broker, value: 'kafka-server:9092'} + - {key: mq.topic, value: spring_test} + refs: + - {parentEndpoint: /case/spring-kafka-case, networkAddress: 'kafka-server:9092', + refType: CrossProcess, parentSpanId: not null, parentTraceSegmentId: not null, + parentServiceInstance: not null, parentService: spring-kafka-2.3.x-scenario, + traceId: not null} \ No newline at end of file diff --git a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java index 5193b9678..03c7f1e5a 100644 --- a/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java +++ b/test/plugin/scenarios/spring-kafka-2.3.x-scenario/src/main/java/test/org/apache/skywalking/apm/testcase/spring/kafka/controller/CaseController.java @@ -41,6 +41,7 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody; import javax.annotation.PostConstruct; +import java.util.Arrays; import java.io.IOException; import java.util.HashMap; import java.util.Map; @@ -57,14 +58,16 @@ public class CaseController { private String bootstrapServers; private String topicName; private KafkaTemplate kafkaTemplate; + private KafkaTemplate kafkaTemplate2; - private CountDownLatch latch = new CountDownLatch(1); + private CountDownLatch latch; private String helloWorld = "helloWorld"; @PostConstruct private void setUp() { topicName = "spring_test"; setUpProvider(); + setUpAnotherProvider(); setUpConsumer(); } @@ -82,6 +85,21 @@ public class CaseController { } } + private void setUpAnotherProvider() { + Map props = new HashMap<>(); + // use list type here + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Arrays.asList(bootstrapServers.split(","))); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + kafkaTemplate2 = new KafkaTemplate(new DefaultKafkaProducerFactory<>(props)); + try { + kafkaTemplate2.send(topicName, "key", "ping").get(); + kafkaTemplate2.flush(); + } catch (Exception e) { + e.printStackTrace(); + } + } + private void setUpConsumer() { Map configs = new HashMap<>(); configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); @@ -115,9 +133,14 @@ public class CaseController { @RequestMapping("/spring-kafka-case") @ResponseBody public String springKafkaCase() throws Exception { + this.latch = new CountDownLatch(1); kafkaTemplate.send(topicName, "key", helloWorld).get(); - latch.await(); + this.latch.await(); kafkaTemplate.flush(); + this.latch = new CountDownLatch(1); + kafkaTemplate2.send(topicName, "key", helloWorld).get(); + this.latch.await(); + kafkaTemplate2.flush(); return SUCCESS; }