fix persistent store

birdayz · Jun 25, 2021 · 4f91544 · 4f91544
1 parent 6587f4f
commit 4f91544
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 4 deletions.
diff --git a/src/main/java/de/nerden/kafka/streams/MoreTransformers.java b/src/main/java/de/nerden/kafka/streams/MoreTransformers.java
@@ -85,17 +85,17 @@ public static <K, V> TransformerSupplier<K, V, KeyValue<K, List<V>>> Batch(
     final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materializedInternal =
         new MaterializedInternal<>(materialized);
 
-    if (materializedInternal.storeName() != null) {
+    if (materializedInternal.storeSupplier() != null) {
       return new BatchTransformerSupplier<>(
-          materializedInternal.storeName(),
+          (KeyValueBytesStoreSupplier) materializedInternal.storeSupplier(),
           materializedInternal.keySerde(),
           materializedInternal.valueSerde(),
           materializedInternal.loggingEnabled(),
           maxBatchDurationMillis,
           maxBatchSizePerKey);
-    } else if (materializedInternal.storeSupplier() != null) {
+    } else if (materializedInternal.storeName() != null) {
       return new BatchTransformerSupplier<>(
-          (KeyValueBytesStoreSupplier) materializedInternal.storeSupplier(),
+          materializedInternal.storeName(),
           materializedInternal.keySerde(),
           materializedInternal.valueSerde(),
           materializedInternal.loggingEnabled(),

diff --git a/src/main/java/de/nerden/kafka/streams/processor/BatchTransformerSupplier.java b/src/main/java/de/nerden/kafka/streams/processor/BatchTransformerSupplier.java
@@ -36,6 +36,7 @@ public BatchTransformerSupplier(
     this.changeLoggingEnabled = changeLoggingEnabled;
     this.maxBatchDurationMillis = maxBatchDurationMillis;
     this.maxBatchSizePerKey = maxBatchSizePerKey;
+    this.storeName = this.storeSupplier.name();
   }
 
   public BatchTransformerSupplier(