add support for Kafka Streams v2.5.0

streamthoughts · May 11, 2020 · eb9d558 · eb9d558
1 parent 6dbed7c
commit eb9d558
Show file tree

Hide file tree

Showing 9 changed files with 32 additions and 7 deletions.
diff --git a/azkarra-api/pom.xml b/azkarra-api/pom.xml
@@ -80,6 +80,7 @@
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-streams</artifactId>
+            <scope>provided</scope>
         </dependency>
 
         <dependency>

diff --git a/azkarra-api/src/main/java/io/streamthoughts/azkarra/api/streams/KafkaStreamsContainer.java b/azkarra-api/src/main/java/io/streamthoughts/azkarra/api/streams/KafkaStreamsContainer.java
@@ -519,8 +519,14 @@ Logger logger() {
         return LOG;
     }
 
+    /**
+     * Checks if the {@link KafkaStreams} is neither RUNNING nor REBALANCING.
+     */
     public boolean isNotRunning() {
-        return !kafkaStreams.state().isRunning();
+        // This is equivalent to the KafkaStreams methods :
+        // State.isRunning() <= 2.4 or State.isRunningOrRebalancing() >= 2.5
+        final KafkaStreams.State state = kafkaStreams.state();
+        return !(state.equals(KafkaStreams.State.RUNNING) || state.equals(KafkaStreams.State.REBALANCING));
     }
 
     /**

diff --git a/azkarra-archetype/src/main/resources/archetype-resources/pom.xml b/azkarra-archetype/src/main/resources/archetype-resources/pom.xml
@@ -41,7 +41,7 @@
     </licenses>
 
     <properties>
-        <kafka.streams.version>2.4.0</kafka.streams.version>
+        <kafka.streams.version>2.5.0</kafka.streams.version>
         <log4j.version>2.12.1</log4j.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
@@ -111,8 +111,13 @@
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-streams</artifactId>
-            <version>2.3.0</version>
             <scope>compile</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-api</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>

diff --git a/azkarra-examples/pom.xml b/azkarra-examples/pom.xml
@@ -80,7 +80,7 @@
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-streams</artifactId>
-            <version>${kafka.streams.version}</version>
+            <scope>compile</scope>
         </dependency>
 
         <dependency>

diff --git a/azkarra-json-serializers/pom.xml b/azkarra-json-serializers/pom.xml
@@ -47,7 +47,6 @@
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-streams</artifactId>
-            <scope>provided</scope>
         </dependency>
 
         <dependency>

diff --git a/azkarra-metrics/pom.xml b/azkarra-metrics/pom.xml
@@ -62,6 +62,11 @@
             <version>1</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-streams</artifactId>
+        </dependency>
+
     </dependencies>
 
 </project>
diff --git a/azkarra-server/pom.xml b/azkarra-server/pom.xml
@@ -228,6 +228,12 @@
             <artifactId>mockito-all</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-streams</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <!-- END dependencies for testing-->
     </dependencies>
 </project>
diff --git a/...-server/src/test/java/io/streamthoughts/azkarra/http/query/HttpRemoteQueryClientTest.java b/...-server/src/test/java/io/streamthoughts/azkarra/http/query/HttpRemoteQueryClientTest.java
@@ -42,6 +42,7 @@
 
 import java.io.IOException;
 import java.net.HttpURLConnection;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
@@ -105,13 +106,14 @@ public void shouldQueryRemoteServerSuccessfully() throws ExecutionException, Int
     }
 
     private QueryResult<String, String> newQueryResult() {
+        List<KV<String, String>> kv = singletonList(KV.of("k1", "v1"));
         return QueryResultBuilder.<String, String>newBuilder()
             .setServer(SERVER_INFO.hostAndPort())
             .setStatus(QueryStatus.SUCCESS)
             .setStoreName(TEST_STORE_NAME)
             .setSuccessResultSet(
                 singletonList(
-                    new SuccessResultSet<>(SERVER_INFO.hostAndPort(), true, singletonList(KV.of("k1", "v1")))
+                    new SuccessResultSet<>(SERVER_INFO.hostAndPort(), true, kv)
                 )
             )
             .setTook(0)

diff --git a/pom.xml b/pom.xml
@@ -85,7 +85,7 @@
     </distributionManagement>
 
     <properties>
-        <kafka.streams.version>2.4.0</kafka.streams.version>
+        <kafka.streams.version>2.5.0</kafka.streams.version>
         <slf4j.version>1.7.28</slf4j.version>
         <log4j.version>2.12.1</log4j.version>
         <junit.version>4.12</junit.version>
@@ -248,6 +248,7 @@
                 <groupId>org.apache.kafka</groupId>
                 <artifactId>kafka-streams</artifactId>
                 <version>${kafka.streams.version}</version>
+                <scope>provided</scope>
             </dependency>
 
             <dependency>