Re: [PR] [INLONG-9225][Audit] Automatically create audit topic after service startup [inlong]

["healchow (via GitHub)" <gi...@apache.org>]

healchow commented on code in PR #9226:
URL: https://github.com/apache/inlong/pull/9226#discussion_r1387562348


##########
inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/service/consume/KafkaConsume.java:
##########
@@ -67,16 +77,70 @@ public void start() {
         Preconditions.checkArgument(StringUtils.isNotEmpty(mqConfig.getKafkaConsumerName()),
                 "no kafka consume name specified");
 
+        // create topic if need
+        createTopic();
+
         initConsumer(mqConfig);
 
         Thread thread = new Thread(new Fetcher(consumer, topic, isAutoCommit, mqConfig.getFetchWaitMs()),
                 "KafkaConsume_Fetcher_Thread");
         thread.start();
     }
 
+    /**
+     * create topic if need
+     */
+    private void createTopic() {
+        int defaultNumPartitions = 3;
+        int defaultReplicationFactor = 2;
+
+        if (StringUtils.isNotEmpty(mqConfig.getNumPartitions())) {
+            defaultNumPartitions = Integer.parseInt(mqConfig.getNumPartitions());
+        }
+
+        if (StringUtils.isNotEmpty(mqConfig.getReplicationFactor())) {
+            defaultReplicationFactor = Integer.parseInt(mqConfig.getReplicationFactor());
+        }
+
+        try (AdminClient adminClient = AdminClient.create(getProperties(mqConfig))) {
+            ListTopicsResult topicList = adminClient.listTopics();
+            KafkaFuture<Set<String>> kafkaFuture = topicList.names();
+            Set<String> topicSet = kafkaFuture.get();
+
+            if (topicSet.contains(topic)) {
+                // not need
+                LOG.info("The audit topic:{} already exists.", topic);
+                return;
+            }
+
+            DescribeClusterResult describeClusterResult = adminClient.describeCluster();
+            Collection<Node> nodes = describeClusterResult.nodes().get();
+            if (nodes.isEmpty())
+                throw new IllegalArgumentException("kafka server not find");
+
+            int partition = Math.min(defaultNumPartitions, nodes.size());
+            int factor = nodes.size() > 1 ? Math.min(defaultReplicationFactor, nodes.size()) : 1;
+
+            NewTopic needCreateTopic = new NewTopic(topic, partition, (short) factor);
+
+            CreateTopicsResult createTopicsResult =
+                    adminClient.createTopics(Collections.singletonList(needCreateTopic));
+            createTopicsResult.all().get();
+
+        } catch (ExecutionException | InterruptedException e) {
+            throw new RuntimeException(
+                    String.format("create audit topic:{} error with config:%s", getProperties(mqConfig)));

Review Comment:
   The `{}` seems not used in here, and suggested print the stacktrace into the log.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org