Implemented TopicAdminService deleteTopic and getTopic. Added tests to cover the methods.

pom.xml

@@ -101,7 +101,6 @@
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
         </dependency>
-
         <dependency>
             <groupId>io.swagger.parser.v3</groupId>
             <artifactId>swagger-parser</artifactId>
@@ -178,8 +177,8 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>3.8.1</version>
                 <configuration>
-                    <source>1.8</source>
-                    <target>1.8</target>
+                    <source>11</source>
+                    <target>11</target>
                     <verbose>false</verbose>
                 </configuration>
             </plugin>

src/main/java/org/opengroup/osdu/streaming/exception/StreamAdminException.java

+package org.opengroup.osdu.streaming.exception;
+
+public class StreamAdminException extends RuntimeException {
+
+    public StreamAdminException(String message) {
+        super(message);
+    }
+
+    public StreamAdminException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public StreamAdminException(Throwable cause) {
+        super(cause);
+    }
+
+}

src/main/java/org/opengroup/osdu/streaming/service/StreamingAdminServiceImpl.java

@@ -16,7 +16,6 @@ package org.opengroup.osdu.streaming.service;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.kafka.clients.admin.NewTopic;
 import org.opengroup.osdu.core.common.model.http.DpsHeaders;
 import org.opengroup.osdu.core.common.model.storage.Record;
 import org.opengroup.osdu.core.common.model.storage.StorageException;
@@ -108,7 +107,7 @@ public class StreamingAdminServiceImpl implements StreamingAdminService {
             upRec = storageService.upsertRecord(rec);
 
             // create topic
-            NewTopic newTopic = this.topicAdminService.createTopic(streamRecord);
+            this.topicAdminService.createTopic(streamRecord);
         } catch (StorageException e) {
             logger.error("Got exception: " + e.getMessage() + "\nFull HTTP Response:" + e.getHttpResponse());
         } catch (JsonProcessingException e) {

src/main/java/org/opengroup/osdu/streaming/service/TopicAdminService.java

 package org.opengroup.osdu.streaming.service;
 
-import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
 import org.opengroup.osdu.streaming.model.StreamRecord;
 
+import java.util.Optional;
+
 public interface TopicAdminService {
 
-    NewTopic createTopic(StreamRecord streamRecord);
+    Optional<TopicDescription> getTopic(StreamRecord streamRecord);
+
+    TopicDescription createTopic(StreamRecord streamRecord);
+
+    void deleteTopic(StreamRecord streamRecord);
 
 }

src/main/java/org/opengroup/osdu/streaming/service/TopicAdminServiceImpl.java

 package org.opengroup.osdu.streaming.service;
 
+import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.opengroup.osdu.streaming.exception.StreamAdminException;
 import org.opengroup.osdu.streaming.model.StreamRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.kafka.config.TopicBuilder;
 import org.springframework.stereotype.Service;
 
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
 @Service
 public class TopicAdminServiceImpl implements TopicAdminService {
 
     private static final Logger logger = LoggerFactory.getLogger(TopicAdminService.class);
 
+    private AdminClient adminClient;
+
+    public TopicAdminServiceImpl(AdminClient adminClient) {
+        this.adminClient = adminClient;
+    }
+
+    @Override
+    public Optional<TopicDescription> getTopic(StreamRecord streamRecord) {
+        try {
+            Map<String, KafkaFuture<TopicDescription>> kafkaFutureMap = adminClient.describeTopics(List.of(streamRecord.getKind())).values();
+            return kafkaFutureMap.containsKey(streamRecord.getKind()) ? Optional.of(kafkaFutureMap.get(streamRecord.getKind()).get()) : Optional.empty();
+        } catch (ExecutionException e) {
+            if (e.getCause() instanceof UnknownTopicOrPartitionException) {
+                return Optional.empty();
+            }
+            throw new StreamAdminException(e);
+        } catch (InterruptedException e) {
+            throw new StreamAdminException(e);
+        }
+    }
+
     @Override
-    public NewTopic createTopic(StreamRecord streamRecord) {
+    public TopicDescription createTopic(StreamRecord streamRecord) {
         logger.debug("Creating Topic for StreamRecord: {}", streamRecord);
-        return TopicBuilder.name(streamRecord.getKind())
-                //TODO: How to setup partitions and topics
-                .partitions(10)
-                .replicas(3)
-                .build();
+        Optional<TopicDescription> optional = this.getTopic(streamRecord);
+        if (optional.isPresent()){
+            return optional.get();
+        }
+        try {
+            NewTopic newTopic = TopicBuilder.name(streamRecord.getKind())
+                    //TODO: How to setup partitions and replicas
+                    //.partitions(10)
+                    //.replicas(3)
+                    .build();
+            adminClient.createTopics(List.of(newTopic)).all().get();
+            return getTopic(streamRecord).get();
+        } catch (InterruptedException | ExecutionException e) {
+            throw new StreamAdminException(e);
+        }
+    }
+
+    @Override
+    public void deleteTopic(StreamRecord streamRecord) {
+        adminClient.deleteTopics(List.of(streamRecord.getKind()));
     }
 
 }

src/main/java/org/opengroup/osdu/streaming/util/KafkaTopicConfiguration.java → src/main/java/org/opengroup/osdu/streaming/util/KafkaAdminConfiguration.java

 package org.opengroup.osdu.streaming.util;
 
+import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -8,9 +10,10 @@ import org.springframework.kafka.core.KafkaAdmin;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 
 @Configuration
-public class KafkaTopicConfiguration {
+public class KafkaAdminConfiguration {
 
     @Value(value = "${kafka.bootstrapAddress:localhost:9092}")
     private String bootstrapAddress;
@@ -19,7 +22,16 @@ public class KafkaTopicConfiguration {
     public KafkaAdmin kafkaAdmin() {
         Map<String, Object> configs = new HashMap<>();
         configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
-        return new KafkaAdmin(configs);
+        KafkaAdmin kafkaAdmin = new KafkaAdmin(configs);
+        return kafkaAdmin;
+    }
+
+    @Bean
+    public AdminClient adminClient() throws ExecutionException, InterruptedException {
+        Map<String, Object> configs = new HashMap<>();
+        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        AdminClient adminClient = KafkaAdminClient.create(configs);
+        return adminClient;
     }
 
 }

src/main/resources/application.properties

@@ -2,4 +2,5 @@
 server.servlet.contextPath=/api/streaming/v1/
 
 # Profile
-spring.profiles.active=local
+spring.profiles.active=local
\ No newline at end of file
+spring.main.allow-bean-definition-overriding=true
\ No newline at end of file

src/test/java/org/opengroup/osdu/streaming/service/TopicAdminServiceTest.java

 package org.opengroup.osdu.streaming.service;
 
-import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
+import org.apache.kafka.clients.admin.TopicDescription;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.opengroup.osdu.streaming.model.StreamRecord;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.core.KafkaAdmin;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.utility.DockerImageName;
 
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 @RunWith(SpringRunner.class)
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@DirtiesContext
 public class TopicAdminServiceTest {
 
     @ClassRule
-    public static KafkaContainer kafka =
-            new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"));
+    public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"));
 
     @Autowired
     private TopicAdminService topicAdminService;
 
     @Test
-    public void topicCreation() {
+    public void topicCreationAndRetrieval() {
+        StreamRecord streamRecord = new StreamRecord();
+        streamRecord.setKind(UUID.randomUUID().toString());
+        TopicDescription topicDescription = topicAdminService.createTopic(streamRecord);
+        assertThat(topicDescription).isNotNull();
+        Optional<TopicDescription> optionalTopicDescription = topicAdminService.getTopic(streamRecord);
+        assertThat(optionalTopicDescription.isPresent()).isTrue();
+    }
+
+    @Test
+    public void topicCreationAndDeletion() {
+        StreamRecord streamRecord = new StreamRecord();
+        streamRecord.setKind(UUID.randomUUID().toString());
+        TopicDescription topicDescription = topicAdminService.createTopic(streamRecord);
+        assertThat(topicDescription).isNotNull();
+        topicAdminService.deleteTopic(streamRecord);
+        Optional<TopicDescription> optionalTopicDescription = topicAdminService.getTopic(streamRecord);
+        assertThat(optionalTopicDescription.isEmpty()).isTrue();
+    }
+
+    @Test
+    public void topicCreateSameName() {
         StreamRecord streamRecord = new StreamRecord();
-        streamRecord.setKind("test");
-        NewTopic newTopic = topicAdminService.createTopic(streamRecord);
-        assertThat(newTopic).isNotNull();
-        assertThat(newTopic.name()).isEqualTo("test");
+        streamRecord.setKind(UUID.randomUUID().toString());
+        TopicDescription topicDescription = topicAdminService.createTopic(streamRecord);
+        assertThat(topicDescription).isNotNull();
+        TopicDescription topicDescription2 = topicAdminService.createTopic(streamRecord);
+        assertThat(topicDescription).isEqualTo(topicDescription2);
+    }
+
+    @TestConfiguration
+    static class KafkaTestContainersConfiguration {
+
+        @Bean
+        public KafkaAdmin kafkaAdmin() {
+            Map<String, Object> configs = new HashMap<>();
+            configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
+            KafkaAdmin kafkaAdmin = new KafkaAdmin(configs);
+            return kafkaAdmin;
+        }
+
+        @Bean
+        public AdminClient adminClient() {
+            Map<String, Object> configs = new HashMap<>();
+            configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
+            AdminClient adminClient = KafkaAdminClient.create(configs);
+            return adminClient;
+        }
+
     }
 
 }