kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add kafka config for how long does kafka query server topic will trigger timeout (#5554)

This commit is contained in:
zifeihan 2020-09-25 13:03:51 +08:00 committed by GitHub
parent 5b5a122b2c
commit 84d49c4412
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 23 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaProducerManager.java
View File

@ -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<String, Bytes> producer;
@Override
@ -60,9 +66,13 @@ public class KafkaProducerManager implements BootService, Runnable {
Set<String> 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();
})

5
apm-sniffer/optional-reporter-plugins/kafka-reporter-plugin/src/main/java/org/apache/skywalking/apm/agent/core/kafka/KafkaReporterPluginConfig.java
View File

@ -43,6 +43,11 @@ public class KafkaReporterPluginConfig {
public static String TOPIC_METER = "skywalking-meters";
public static Map<String, String> PRODUCER_CONFIG = new HashMap<>();
/**
* Timeout period of reading topics from the Kafka server, the unit is second.
*/
public static int GET_TOPIC_TIMEOUT = 10;
}
}
}

1
docs/en/setup/service-agent/java-agent/How-to-enable-kafka-reporter.md
View File

@ -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).

1
docs/en/setup/service-agent/java-agent/README.md
View File

@ -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` |

1
test/e2e/e2e-test/docker/kafka/Dockerfile.provider
View File

@ -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

1
test/e2e/e2e-test/docker/kafka/docker-compose.base.yml
View File

@ -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

1
test/e2e/e2e-test/docker/kafka/docker-compose.meter.yml
View File

@ -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

1
test/e2e/e2e-test/docker/kafka/docker-compose.profiling.yml
View File

@ -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