From 7654161721ab6807da1cf57576fb778ea878fc44 Mon Sep 17 00:00:00 2001 From: Cool-Coding Date: Tue, 20 Jul 2021 08:50:36 +0800 Subject: [PATCH] Fix the problem of Kafka transport topics are created duplicated with and without namespace (#7326) (#7329) --- CHANGES.md | 3 +- .../kafka/KafkaFetcherHandlerRegister.java | 85 +++++++++---------- .../kafka/provider/KafkaFetcherProvider.java | 2 +- 3 files changed, 45 insertions(+), 45 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 52fd3222cc..f32f39882f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -108,9 +108,10 @@ Release Notes. * The `core/syncThreads` setting(added in 8.5.0) is removed due to metrics persistence is fully asynchronous. * Optimization: Concurrency mode of execution stage for metrics is removed(added in 8.5.0). Only concurrency of prepare stage is meaningful and kept. -* Fix -meters metrics topic isn't created with namespace issue +* Fix `-meters` metrics topic isn't created with namespace issue * Enhance persistent session timeout mechanism. Because the enhanced session could cache the metadata metrics forever, new timeout mechanism is designed for avoiding this specific case. +* Fix Kafka transport topics are created duplicated with and without namespace issue #### UI diff --git a/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/KafkaFetcherHandlerRegister.java b/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/KafkaFetcherHandlerRegister.java index f221e73969..03db464550 100644 --- a/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/KafkaFetcherHandlerRegister.java +++ b/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/KafkaFetcherHandlerRegister.java @@ -21,6 +21,7 @@ package org.apache.skywalking.oap.server.analyzer.agent.kafka; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Lists; import java.time.Duration; +import java.util.Collection; import java.util.List; import java.util.Objects; import java.util.Properties; @@ -60,60 +61,21 @@ public class KafkaFetcherHandlerRegister implements Runnable { private KafkaConsumer consumer = null; private final KafkaFetcherConfig config; private final boolean isSharding; + private final Properties properties; private int threadPoolSize = Runtime.getRuntime().availableProcessors() * 2; private int threadPoolQueueSize = 10000; private final ThreadPoolExecutor executor; private final boolean enableKafkaMessageAutoCommit; - public KafkaFetcherHandlerRegister(KafkaFetcherConfig config) throws ModuleStartException { + public KafkaFetcherHandlerRegister(KafkaFetcherConfig config) { this.config = config; - Properties properties = new Properties(); + properties = new Properties(); properties.putAll(config.getKafkaConsumerConfig()); properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, config.getGroupId()); properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); - AdminClient adminClient = AdminClient.create(properties); - Set missedTopics = adminClient.describeTopics(Lists.newArrayList( - config.getTopicNameOfManagements(), - config.getTopicNameOfMetrics(), - config.getTopicNameOfProfiling(), - config.getTopicNameOfTracingSegments(), - config.getTopicNameOfMeters(), - config.getTopicNameOfLogs(), - config.getTopicNameOfJsonLogs() - )) - .values() - .entrySet() - .stream() - .map(entry -> { - try { - entry.getValue().get(); - return null; - } catch (InterruptedException | ExecutionException ignore) { - } - return entry.getKey(); - }) - .filter(Objects::nonNull) - .collect(Collectors.toSet()); - - if (!missedTopics.isEmpty()) { - log.info("Topics" + missedTopics.toString() + " not exist."); - List newTopicList = missedTopics.stream() - .map(topic -> new NewTopic( - topic, - config.getPartitions(), - (short) config.getReplicationFactor() - )).collect(Collectors.toList()); - - try { - adminClient.createTopics(newTopicList).all().get(); - } catch (Exception e) { - throw new ModuleStartException("Failed to create Kafka Topics" + missedTopics + ".", e); - } - } - if (config.isSharding() && StringUtil.isNotEmpty(config.getConsumePartitions())) { isSharding = true; } else { @@ -142,8 +104,11 @@ public class KafkaFetcherHandlerRegister implements Runnable { topicPartitions.addAll(handler.getTopicPartitions()); } - public void start() { + public void start() throws ModuleStartException { handlerMap = builder.build(); + + createTopicIfNeeded(handlerMap.keySet(), properties); + if (isSharding) { consumer.assign(topicPartitions); } else { @@ -171,4 +136,38 @@ public class KafkaFetcherHandlerRegister implements Runnable { } } } + + private void createTopicIfNeeded(Collection topics, Properties properties) throws ModuleStartException { + AdminClient adminClient = AdminClient.create(properties); + Set missedTopics = adminClient.describeTopics(topics) + .values() + .entrySet() + .stream() + .map(entry -> { + try { + entry.getValue().get(); + return null; + } catch (InterruptedException | ExecutionException ignore) { + } + return entry.getKey(); + }) + .filter(Objects::nonNull) + .collect(Collectors.toSet()); + + if (!missedTopics.isEmpty()) { + log.info("Topics" + missedTopics.toString() + " not exist."); + List newTopicList = missedTopics.stream() + .map(topic -> new NewTopic( + topic, + config.getPartitions(), + (short) config.getReplicationFactor() + )).collect(Collectors.toList()); + + try { + adminClient.createTopics(newTopicList).all().get(); + } catch (Exception e) { + throw new ModuleStartException("Failed to create Kafka Topics" + missedTopics + ".", e); + } + } + } } diff --git a/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/KafkaFetcherProvider.java b/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/KafkaFetcherProvider.java index c68a6bc802..c07036ffee 100644 --- a/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/KafkaFetcherProvider.java +++ b/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/KafkaFetcherProvider.java @@ -69,7 +69,7 @@ public class KafkaFetcherProvider extends ModuleProvider { } @Override - public void start() throws ServiceNotProvidedException { + public void start() throws ServiceNotProvidedException, ModuleStartException { handlerRegister.register(new JVMMetricsHandler(getManager(), config)); handlerRegister.register(new ServiceManagementHandler(getManager(), config)); handlerRegister.register(new TraceSegmentHandler(getManager(), config));