[#noissue] Explicitly create topics for kafka-streams integration tests

Author: ga-ram

plugins-it/kafka-it/kafka-3-it/src/test/java/test/pinpoint/plugin/kafka/KafkaStreamsUnitServer.java

@@ -17,23 +17,28 @@
 package test.pinpoint.plugin.kafka;
 
 import com.navercorp.pinpoint.test.plugin.shared.SharedTestLifeCycle;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.streams.errors.StreamsException;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.junit.Assume;
 import org.testcontainers.DockerClientFactory;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.utility.DockerImageName;
 
 import java.util.Properties;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.HashMap;
+
+import static test.pinpoint.plugin.kafka.KafkaITConstants.INPUT_TOPIC;
+import static test.pinpoint.plugin.kafka.KafkaITConstants.OUTPUT_TOPIC;
 
 /**
  * Copy of https://github.com/chbatey/kafka-unit/blob/master/src/main/java/info/batey/kafka/unit/KafkaUnit.java
  * Some codes have been modified for testing from the copied code.
  */
 public class KafkaStreamsUnitServer implements SharedTestLifeCycle {
-    private static final Logger logger = LogManager.getLogger(KafkaStreamsUnitServer.class);
-
 
     private KafkaContainer container;
     private TestStream TEST_STREAM;
@@ -48,6 +53,8 @@ public Properties beforeAll() {
 
         String brokerUrl = "localhost:" + port;
 
+        createTopics(brokerUrl);
+
         TEST_STREAM = new TestStream(brokerUrl);
         TEST_STREAM.start();
         System.out.println();
@@ -57,6 +64,17 @@ public Properties beforeAll() {
         return properties;
     }
 
+    private static void createTopics(String brokerUrl) {
+        List<NewTopic> toCreate = new ArrayList<>();
+        toCreate.add(new NewTopic(INPUT_TOPIC, 1, (short) 1));
+        toCreate.add(new NewTopic(OUTPUT_TOPIC, 1, (short) 1));
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("bootstrap.servers", brokerUrl);
+
+        AdminClient.create(config).createTopics(toCreate);
+    }
+
     @Override
     public void afterAll() {
         if (TEST_STREAM != null) {