Merge branch 'demo/aws' into 'develop'

testing in AWS environment

See merge request !12

Dockerfile

-FROM maven:3-jdk-8-alpine
+FROM openjdk:8-jre
 
-WORKDIR /usr/src/app
+WORKDIR /usr/lib/osdu
 
-COPY . /usr/src/app
-RUN mvn package
+COPY ./target/*.jar /usr/lib/osdu
+#RUN mvn package
 
-ENV PORT 5000
+ENV PORT 8080
 EXPOSE $PORT
-CMD [ "sh", "-c", "mvn -Dserver.port=${PORT} spring-boot:run" ]
+#CMD [ "sh", "-c", "mvn -Dserver.port=${PORT} spring-boot:run" ]
+CMD [ "sh", "java", "-jar", "stream-admin-service-0.0.1-SNAPSHOT.jar" ]

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

@@ -16,12 +16,14 @@ package org.opengroup.osdu.streaming.service;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+
 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;
 import org.opengroup.osdu.core.common.model.storage.UpsertRecords;
 import org.opengroup.osdu.core.common.storage.IStorageFactory;
 import org.opengroup.osdu.core.common.storage.IStorageService;
+import org.opengroup.osdu.streaming.model.StreamDatasetDatasetProperties.StreamTypeEnum;
 import org.opengroup.osdu.streaming.model.StreamRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -47,7 +49,7 @@ public class StreamingAdminServiceImpl implements StreamingAdminService {
     private DeploymentAdminService deploymentAdminService;
 
     public StreamingAdminServiceImpl(ObjectMapper objectMapper, DpsHeaders headers, IStorageFactory storageFactory,
-                                     TopicAdminService topicAdminService, DeploymentAdminService deploymentAdminService) {
+            TopicAdminService topicAdminService, DeploymentAdminService deploymentAdminService) {
         this.objectMapper = objectMapper;
         this.headers = headers;
         this.storageFactory = storageFactory;
@@ -109,10 +111,18 @@ public class StreamingAdminServiceImpl implements StreamingAdminService {
             upRec = storageService.upsertRecord(rec);
 
             // create topic
-            this.topicAdminService.createTopic(streamRecord);
+            // this.topicAdminService.createTopic(streamRecord);
+            StreamTypeEnum streamType = streamRecord.getData().getDatasetProperties().getStreamType();
+            logger.debug("Stream Type: {}", streamType);
+            if (streamType == StreamTypeEnum.SOURCE)
+                this.topicAdminService.createTopics(streamRecord.getData().getDatasetProperties().getStreamDefinition()
+                        .getSinkBindings().stream().toArray(String[]::new));
+
+            // create deployment
             this.deploymentAdminService.createStreamDeployment(streamRecord);
+
         } catch (StorageException e) {
-            logger.error("Got exception: " + e.getMessage() + "\nFull HTTP Response:" + e.getHttpResponse());
+            logger.error("Got exception from storage service: " + e.getMessage() + "\nFull HTTP Response:" + e.getHttpResponse());
         } catch (JsonProcessingException e) {
             logger.error("Got exception: " + e.getMessage() + "\nLocation" + e.getLocation().toString());
         }

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

@@ -11,6 +11,8 @@ public interface TopicAdminService {
 
     TopicDescription createTopic(StreamRecord streamRecord);
 
+    void createTopics(String[] topicList);
+
     void deleteTopic(StreamRecord streamRecord);
 
 }

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

 package org.opengroup.osdu.streaming.service;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
 import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
 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.TopicExistsException;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.opengroup.osdu.streaming.exception.StreamAdminException;
 import org.opengroup.osdu.streaming.model.StreamRecord;
@@ -12,11 +21,6 @@ import org.slf4j.LoggerFactory;
 import org.springframework.kafka.config.TopicBuilder;
 import org.springframework.stereotype.Service;
 
-import java.util.Collections;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.ExecutionException;
-
 @Service
 public class TopicAdminServiceImpl implements TopicAdminService {
 
@@ -31,8 +35,11 @@ public class TopicAdminServiceImpl implements TopicAdminService {
     @Override
     public Optional<TopicDescription> getTopic(StreamRecord streamRecord) {
         try {
-            Map<String, KafkaFuture<TopicDescription>> kafkaFutureMap = adminClient.describeTopics(Collections.singletonList(streamRecord.getKind())).values();
-            return kafkaFutureMap.containsKey(streamRecord.getKind()) ? Optional.of(kafkaFutureMap.get(streamRecord.getKind()).get()) : Optional.empty();
+            Map<String, KafkaFuture<TopicDescription>> kafkaFutureMap = adminClient
+                    .describeTopics(Collections.singletonList(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();
@@ -47,20 +54,21 @@ public class TopicAdminServiceImpl implements TopicAdminService {
     public TopicDescription createTopic(StreamRecord streamRecord) {
         logger.debug("Creating Topic for StreamRecord: {}", streamRecord);
         Optional<TopicDescription> optional = this.getTopic(streamRecord);
-        if (optional.isPresent()){
+        if (optional.isPresent()) {
             return optional.get();
         }
         try {
             NewTopic newTopic = TopicBuilder.name(streamRecord.getKind())
-                    //TODO: How to setup partitions and replicas
-                    //.partitions(10)
-                    //.replicas(3)
+                    // TODO: How to setup partitions and replicas
+                    // .partitions(10)
+                    // .replicas(3)
                     .build();
             adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
             return getTopic(streamRecord).get();
         } catch (InterruptedException | ExecutionException e) {
             throw new StreamAdminException(e);
         }
+
     }
 
     @Override
@@ -68,4 +76,24 @@ public class TopicAdminServiceImpl implements TopicAdminService {
         adminClient.deleteTopics(Collections.singletonList(streamRecord.getKind()));
     }
 
+    @Override
+    public void createTopics(String[] topicList) {
+        logger.debug("Creating Topic(s): {}", topicList);
+        List<NewTopic> newTopics = new ArrayList<NewTopic>();
+        try {
+            for (String topicName : topicList) {
+                NewTopic newTopic = TopicBuilder.name(topicName).build();
+                newTopics.add(newTopic);
+            }
+            CreateTopicsResult topics = adminClient.createTopics(newTopics);
+            logger.debug("Creating Topic(s): {}", topics.all().get());
+        } catch (InterruptedException | ExecutionException e) {
+            if (e.getCause().getClass().equals(TopicExistsException.class)) {
+                logger.warn(e.getMessage());
+            } else {
+                logger.debug("Got exception with cause|class: {}|{}", e.getCause(), e.getClass());
+                throw new StreamAdminException(e);
+            }
+        }
+    }
 }

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

@@ -14,7 +14,7 @@ import java.util.Map;
 @Configuration
 public class KafkaAdminConfiguration {
 
-    @Value(value = "${kafka.bootstrapAddress:localhost:9092}")
+    @Value(value = "${kafka.bootstrapAddress}")
     private String bootstrapAddress;
 
     @Bean

src/main/resources/application-dev_aws.properties

+# OSDU
+osdu.storage.api=http://os-storage:8080/api/storage/v2
+
+# Logging
+logging.level.org.opengroup.osdu.streaming=debug
+# spring web logging level
+logging.level.web=debug
+# this turns on debugging the http requests
+spring.mvc.log-request-details=true
+
+# Kafka
+kafka.bootstrapAddress=kafka.osdu-streams:9092
\ No newline at end of file

src/main/resources/application.properties

@@ -2,12 +2,9 @@
 server.servlet.contextPath=/api/streaming/v1/
 
 # Profile
-spring.profiles.active=local
+spring.profiles.active=dev_aws
 spring.main.allow-bean-definition-overriding=true
 
 logging.level.root=INFO
 
-osdu.storage.api=https://blah:1234/api/storage/v2
-
-
 deployment.namespace=osdu-streams
\ No newline at end of file

src/main/resources/manifests/parser-worker-template.yaml

@@ -4,7 +4,7 @@ metadata:
   name: streams-${deploymentName}-deployment
   namespace: osdu-streams
 spec:
-  replicas: 1
+  replicas: 0
   selector:
     matchLabels:
       run: ${deploymentName}