diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java index 67b3c31135c85..e9291cc514877 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java @@ -295,12 +295,28 @@ private KTable doTransformValues(final ValueTransformerWithKeySuppli final MaterializedInternal> materializedInternal, final String... stateStoreNames) { Objects.requireNonNull(stateStoreNames, "stateStoreNames"); + final Serde keySerde; + final Serde valueSerde; + final String queryableStoreName; + final StoreBuilder> storeBuilder; - final String name = builder.newProcessorName(TRANSFORMVALUES_NAME); + if (materializedInternal != null) { + // don't inherit parent value serde, since this operation may change the value type, more specifically: + // we preserve the key following the order of 1) materialized, 2) parent, 3) null + keySerde = materializedInternal.keySerde() != null ? materializedInternal.keySerde() : this.keySerde; + // we preserve the value following the order of 1) materialized, 2) null + valueSerde = materializedInternal.valueSerde(); + queryableStoreName = materializedInternal.queryableStoreName(); + // only materialize if materialized is specified and it has queryable name + storeBuilder = queryableStoreName != null ? (new KeyValueStoreMaterializer<>(materializedInternal)).materialize() : null; + } else { + keySerde = this.keySerde; + valueSerde = null; + queryableStoreName = null; + storeBuilder = null; + } - // only materialize if users provide a specific queryable name - final String queryableStoreName = materializedInternal != null ? materializedInternal.queryableStoreName() : null; - final StoreBuilder> storeBuilder = queryableStoreName != null ? (new KeyValueStoreMaterializer<>(materializedInternal)).materialize() : null; + final String name = builder.newProcessorName(TRANSFORMVALUES_NAME); final KTableProcessorSupplier processorSupplier = new KTableTransformValues<>( this, @@ -320,13 +336,10 @@ private KTable doTransformValues(final ValueTransformerWithKeySuppli builder.addGraphNode(this.streamsGraphNode, tableNode); - // don't inherit parent value serde, since this operation may change the value type, more specifically: - // we preserve the key following the order of 1) materialized, 2) parent, 3) null - // we preserve the value following the order of 1) materialized, 2) null return new KTableImpl<>( name, - materializedInternal != null && materializedInternal.keySerde() != null ? materializedInternal.keySerde() : keySerde, - materializedInternal != null ? materializedInternal.valueSerde() : null, + keySerde, + valueSerde, sourceNodes, queryableStoreName, processorSupplier,