Merge pull request #39013 from quarkusio/dependabot/maven/kafka3.vers…

…ion-3.7.0 Bump kafka3.version from 3.6.1 to 3.7.0
quarkusio · Mar 5, 2024 · 327dd98 · 327dd98
2 parents a6f9f92 + b9de947
commit 327dd98
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 1 deletion.
diff --git a/bom/application/pom.xml b/bom/application/pom.xml
@@ -148,7 +148,7 @@
         <reactive-streams.version>1.0.4</reactive-streams.version>
         <jboss-logging.version>3.5.3.Final</jboss-logging.version>
         <mutiny.version>2.5.7</mutiny.version>
-        <kafka3.version>3.6.1</kafka3.version>
+        <kafka3.version>3.7.0</kafka3.version>
         <lz4.version>1.8.0</lz4.version> <!-- dependency of the kafka-clients that could be overridden by other imported BOMs in the platform -->
         <snappy.version>1.1.10.5</snappy.version>
         <strimzi-test-container.version>0.100.0</strimzi-test-container.version>

diff --git a/...s/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java b/...s/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
@@ -96,13 +96,30 @@ private void registerCompulsoryClasses(BuildProducer<ReflectiveClassBuildItem> r
         reflectiveClasses.produce(ReflectiveClassBuildItem
                 .builder("org.apache.kafka.streams.processor.internals.StateDirectory$StateDirectoryProcessFile")
                 .methods().fields().build());
+
+        // Listed in BuiltInDslStoreSuppliers
+        reflectiveClasses.produce(ReflectiveClassBuildItem
+                .builder(org.apache.kafka.streams.state.BuiltInDslStoreSuppliers.RocksDBDslStoreSuppliers.class,
+                        org.apache.kafka.streams.state.BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class)
+                .build());
     }
 
     private void registerClassesThatClientMaySpecify(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
             LaunchModeBuildItem launchMode) {
         Properties properties = buildKafkaStreamsProperties(launchMode.getLaunchMode());
         registerExceptionHandler(reflectiveClasses, properties);
         registerDefaultSerdes(reflectiveClasses, properties);
+        registerDslStoreSupplier(reflectiveClasses, properties);
+    }
+
+    private void registerDslStoreSupplier(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
+            Properties kafkaStreamsProperties) {
+        String dlsStoreSupplierClassName = kafkaStreamsProperties
+                .getProperty(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG);
+
+        if (dlsStoreSupplierClassName != null) {
+            registerClassName(reflectiveClasses, dlsStoreSupplierClassName);
+        }
     }
 
     private void registerExceptionHandler(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,