Merge pull request #3 from CDOT-CV/deduplicator--ode-4.0-compatibility

Updated Deduplicator for Compatibility with ode 4.0

Author: John-Wiens

README.md

@@ -205,9 +205,7 @@ The JPO-Deduplicator is a Kafka Java spring-boot application designed to reduce
 
 ### Deduplication Config
 
-When running the jpo-deduplication as a submodule in jpo-utils, the deduplicator will automatically turn on deduplication for a topic when that topic is created. For example if the KAFKA_TOPIC_CREATE_GEOJSONCONVERTER environment variable is set to true, the deduplicator will start performing deduplication for ProcessedMap, ProcessedMapWKT, and ProcessedSpat data. 
-
-To manually configure deduplication for a topic, the following environment variables can also be used.
+When running the jpo-deduplication as a submodule in jpo-utils, the deduplicator will automatically configure an algorithm as enabled or disabled depending on if the corresponding subcomponent is also active.  For example if the KAFKA_TOPIC_CREATE_GEOJSONCONVERTER environment variable is set to true, the deduplicator will start performing deduplication for ProcessedMap, ProcessedMapWKT, and ProcessedSpat data. If the KAFKA_TOPIC_CREATE_GEOJSONCONVERTER is set to false, the deduplicator will disable deduplication for those same topics. To manually configure deduplication for a topic, the following environment variables can also be used. If no value is passed for a given environment variable, the corresponding deduplication algorithm will default to enabled.
 
 | Environment Variable | Description |
 |---|---|

jpo-deduplicator/jpo-deduplicator/pom.xml

@@ -101,23 +101,23 @@
         <dependency>
             <groupId>usdot.jpo.ode</groupId>
             <artifactId>jpo-ode-core</artifactId>
-            <version>3.0.0</version>
+            <version>4.0.0</version>
         </dependency>
         <dependency>
             <groupId>usdot.jpo.ode</groupId>
             <artifactId>jpo-ode-plugins</artifactId>
-            <version>3.0.0</version>
+            <version>4.0.0</version>
         </dependency>
         <dependency>
             <groupId>usdot.jpo.ode</groupId>
             <artifactId>jpo-geojsonconverter</artifactId>
-            <version>1.4.2</version>
+            <version>2.0.0</version>
         </dependency>
         <dependency>
             <groupId>usdot.jpo.ode</groupId>
             <artifactId>jpo-conflictmonitor</artifactId>
             <version>1.5.0</version>
-        </dependency> 
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

jpo-deduplicator/jpo-deduplicator/src/main/java/us/dot/its/jpo/deduplicator/deduplicator/topologies/ProcessedSpatDeduplicatorTopology.java

@@ -66,8 +66,6 @@ public Topology buildTopology() {
                 Serdes.String(), JsonSerdes.ProcessedSpat()));
 
         KStream<String, ProcessedSpat> deduplicatedStream = inputStream.process(new ProcessedSpatProcessorSupplier(props), props.getKafkaStateStoreProcessedSpatName());
-        
-        deduplicatedStream.print(Printed.toSysOut());
 
         deduplicatedStream.to(props.getKafkaTopicDeduplicatedProcessedSpat(), Produced.with(Serdes.String(), JsonSerdes.ProcessedSpat()));
 

jpo-deduplicator/jpo-deduplicator/src/main/resources/application.yaml

@@ -17,27 +17,27 @@ log4j.logger.org.apache.kafka: OFF
 # Processed Map Configuration
 kafkaTopicProcessedMap: topic.ProcessedMap
 kafkaTopicDeduplicatedProcessedMap: topic.DeduplicatedProcessedMap
-enableProcessedMapDeduplication: false
+enableProcessedMapDeduplication: true
 
 # Processed Map WKT Configuration
 kafkaTopicProcessedMapWKT: topic.ProcessedMapWKT
 kafkaTopicDeduplicatedProcessedMapWKT: topic.DeduplicatedProcessedMapWKT
-enableProcessedMapWktDeduplication: false
+enableProcessedMapWktDeduplication: true
 
 # Ode Map Json Configuration
 kafkaTopicOdeMapJson: topic.OdeMapJson
 kafkaTopicDeduplicatedOdeMapJson: topic.DeduplicatedOdeMapJson
-enableOdeMapDeduplication: false
+enableOdeMapDeduplication: true
 
 # Ode Tim Json Configuration
 kafkaTopicOdeTimJson: topic.OdeTimJson
 kafkaTopicDeduplicatedOdeTimJson: topic.DeduplicatedOdeTimJson
-enableOdeTimDeduplication: false
+enableOdeTimDeduplication: true
 
 # Ode Raw Encoded Tim Json Configuration
 kafkaTopicOdeRawEncodedTimJson: topic.OdeRawEncodedTIMJson
 kafkaTopicDeduplicatedOdeRawEncodedTimJson: topic.DeduplicatedOdeRawEncodedTIMJson
-enableOdeRawEncodedTimDeduplication: false
+enableOdeRawEncodedTimDeduplication: true
 
 # Ode Bsm Json Configuration
 kafkaTopicOdeBsmJson: topic.OdeBsmJson
@@ -50,7 +50,7 @@ odeBsmAlwaysIncludeAtSpeed: 1 # Meter / Second
 # Processed Map Configuration
 kafkaTopicProcessedSpat: topic.ProcessedSpat
 kafkaTopicDeduplicatedProcessedSpat: topic.DeduplicatedProcessedSpat
-enableProcessedSpatDeduplication: false
+enableProcessedSpatDeduplication: true
 
 
 # Amount of time to wait to try and increase batching