diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java index 135abec9c0..70ebcf6a18 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java @@ -23,6 +23,8 @@ import java.util.Objects; import java.util.Properties; import java.util.Set; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.DescribeTopicsResult; @@ -33,6 +35,8 @@ import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.common.utils.Bytes; import org.apache.skywalking.apm.agent.core.boot.BootService; import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor; +import org.apache.skywalking.apm.agent.core.logging.api.ILog; +import org.apache.skywalking.apm.agent.core.logging.api.LogManager; /** * Configuring, initializing and holding a KafkaProducer instance for reporters. @@ -40,6 +44,8 @@ import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor; @DefaultImplementor public class KafkaProducerManager implements BootService, Runnable { + private static final ILog LOGGER = LogManager.getLogger(KafkaProducerManager.class); + private KafkaProducer producer; @Override @@ -60,9 +66,13 @@ public class KafkaProducerManager implements BootService, Runnable { Set topics = topicsResult.values().entrySet().stream() .map(entry -> { try { - entry.getValue().get(); + entry.getValue().get( + KafkaReporterPluginConfig.Plugin.Kafka.GET_TOPIC_TIMEOUT, + TimeUnit.SECONDS + ); return null; - } catch (InterruptedException | ExecutionException e) { + } catch (InterruptedException | ExecutionException | TimeoutException e) { + LOGGER.error(e, "Get KAFKA topic:{} error.", entry.getKey()); } return entry.getKey(); }) diff --git a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaReporterPluginConfig.java b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaReporterPluginConfig.java index 7dd693f05b..e401bde353 100644 --- a/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaReporterPluginConfig.java +++ b/apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaReporterPluginConfig.java @@ -43,6 +43,11 @@ public class KafkaReporterPluginConfig { public static String TOPIC_METER = "skywalking-meters"; public static Map PRODUCER_CONFIG = new HashMap<>(); + + /** + * Timeout period of reading topics from the Kafka server, the unit is second. + */ + public static int GET_TOPIC_TIMEOUT = 10; } } } diff --git a/docs/en/setup/service-agent/java-agent/How-to-enable-kafka-reporter.md b/docs/en/setup/service-agent/java-agent/How-to-enable-kafka-reporter.md index 0d2fbe586f..7e7b8245aa 100644 --- a/docs/en/setup/service-agent/java-agent/How-to-enable-kafka-reporter.md +++ b/docs/en/setup/service-agent/java-agent/How-to-enable-kafka-reporter.md @@ -11,6 +11,7 @@ collector.backend_service=${SW_AGENT_COLLECTOR_BACKEND_SERVICES:127.0.0.1:11800} # Kafka producer configuration plugin.kafka.bootstrap_servers=${SW_KAFKA_BOOTSTRAP_SERVERS:localhost:9092} plugin.kafka.producer_config[delivery.timeout.ms]=12000 +plugin.kafka.get_topic_timeout=${SW_GET_TOPIC_TIMEOUT:10} ``` Kafka reporter plugin support to customize all configurations of listed in [here](http://kafka.apache.org/24/documentation.html#producerconfigs). diff --git a/docs/en/setup/service-agent/java-agent/README.md b/docs/en/setup/service-agent/java-agent/README.md index 54ae20d0d3..970fa984b1 100755 --- a/docs/en/setup/service-agent/java-agent/README.md +++ b/docs/en/setup/service-agent/java-agent/README.md @@ -149,6 +149,7 @@ property key | Description | Default | `plugin.dubbo.collect_provider_arguments`| Apache Dubbo provider collect `arguments` in RPC call, use `Object#toString` to collect `arguments`. |`false`| `plugin.dubbo.consumer_provider_length_threshold`| When `plugin.dubbo.provider_consumer_arguments` is `true`, Arguments of length from the front will to the OAP backend |`256`| `plugin.kafka.bootstrap_servers`| A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. | `localhost:9092` +`plugin.kafka.get_topic_timeout`| Timeout period of reading topics from the Kafka server, the unit is second. |`10`| `plugin.kafka.consumer_config`| Kafka producer configuration. || `plugin.kafka.producer_config`| Kafka producer configuration. Read [producer configure](http://kafka.apache.org/24/documentation.html#producerconfigs) to get more details. Check [Kafka report doc](How-to-enable-kafka-reporter.md) for more details and examples. | | `plugin.kafka.topic_meter` | Specify which Kafka topic name for Meter System data to report to. | `skywalking_meters` | diff --git a/test/e2e/e2e-test/docker/kafka/Dockerfile.provider b/test/e2e/e2e-test/docker/kafka/Dockerfile.provider index 142278d3ec..621bf8fde6 100644 --- a/test/e2e/e2e-test/docker/kafka/Dockerfile.provider +++ b/test/e2e/e2e-test/docker/kafka/Dockerfile.provider @@ -26,6 +26,7 @@ RUN tar -zxf apache-skywalking-apm*.gz --strip=1 || true RUN mv apache-skywalking-apm-bin/* . || true RUN sed -i 's/# plugin.kafka.bootstrap_servers/plugin.kafka.bootstrap_servers/g' agent/config/agent.config +RUN echo 'plugin.kafka.get_topic_timeout=${SW_GET_TOPIC_TIMEOUT:60}' >> agent/config/agent.config # activate kafka-reporter plugin RUN mv agent/optional-reporter-plugins/kafka-reporter-plugin-*.jar agent/plugins/ || true diff --git a/test/e2e/e2e-test/docker/kafka/docker-compose.base.yml b/test/e2e/e2e-test/docker/kafka/docker-compose.base.yml index 2b8652b137..baa7e63d00 100644 --- a/test/e2e/e2e-test/docker/kafka/docker-compose.base.yml +++ b/test/e2e/e2e-test/docker/kafka/docker-compose.base.yml @@ -49,6 +49,7 @@ services: environment: SW_AGENT_COLLECTOR_BACKEND_SERVICES: oap:11800 SW_KAFKA_BOOTSTRAP_SERVERS: broker-a:9092,broker-b:9092 + SW_GET_TOPIC_TIMEOUT: 60 JAVA_OPTS: >- -javaagent:/jacoco/jacocoagent.jar=classdumpdir=/jacoco/classes/provider,destfile=/jacoco/provider.exec,includes=org.apache.skywalking.*,excludes=org.apache.skywalking.apm.dependencies.* -javaagent:/skywalking/agent/skywalking-agent.jar=logging.output=CONSOLE diff --git a/test/e2e/e2e-test/docker/kafka/docker-compose.meter.yml b/test/e2e/e2e-test/docker/kafka/docker-compose.meter.yml index da4577fbe7..c3f2240523 100644 --- a/test/e2e/e2e-test/docker/kafka/docker-compose.meter.yml +++ b/test/e2e/e2e-test/docker/kafka/docker-compose.meter.yml @@ -48,6 +48,7 @@ services: environment: SW_AGENT_COLLECTOR_BACKEND_SERVICES: oap:11800 SW_KAFKA_BOOTSTRAP_SERVERS: broker-a:9092,broker-b:9092 + SW_GET_TOPIC_TIMEOUT: 60 JAVA_OPTS: >- -javaagent:/jacoco/jacocoagent.jar=classdumpdir=/jacoco/classes/provider,destfile=/jacoco/provider.exec,includes=org.apache.skywalking.*,excludes=org.apache.skywalking.apm.dependencies.* -javaagent:/skywalking/agent/skywalking-agent.jar=logging.output=CONSOLE diff --git a/test/e2e/e2e-test/docker/kafka/docker-compose.profiling.yml b/test/e2e/e2e-test/docker/kafka/docker-compose.profiling.yml index 4b232edc00..2c414ac6b5 100644 --- a/test/e2e/e2e-test/docker/kafka/docker-compose.profiling.yml +++ b/test/e2e/e2e-test/docker/kafka/docker-compose.profiling.yml @@ -62,6 +62,7 @@ services: SW_AGENT_NAME: e2e-profile-service SW_AGENT_COLLECTOR_BACKEND_SERVICES: oap:11800 SW_KAFKA_BOOTSTRAP_SERVERS: broker-a:9092,broker-b:9092 + SW_GET_TOPIC_TIMEOUT: 60 JAVA_OPTS: >- -javaagent:/jacoco/jacocoagent.jar=classdumpdir=/jacoco/classes/provider,destfile=/jacoco/provider.exec,includes=org.apache.skywalking.*,excludes=org.apache.skywalking.apm.dependencies.* -javaagent:/skywalking/agent/skywalking-agent.jar=logging.output=CONSOLE