private void createTopicIfNeeded(Collection<String> topics, Properties properties) throws ModuleStartException {
        Properties adminProps = new Properties();
        adminProps.putAll(properties);
        // remove 'group.id' to avoid unknown configure warning.
        adminProps.remove(ConsumerConfig.GROUP_ID_CONFIG);

        AdminClient adminClient = AdminClient.create(adminProps);
        Set<String> 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 + " not exist.");
            List<NewTopic> 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);
            }
        }
    }
Configuring and initializing a KafkaConsumer client as a dispatcher to delivery Kafka Message to registered handler
by topic.
oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/KafkaFetcherHandlerRegister.java