KStreamImpl.doTransformValues

Author: jaceklaskowski

SUMMARY.adoc

@@ -161,6 +161,7 @@
 .. link:kafka-streams-internals-TableSourceNode.adoc[TableSourceNode]
 .. link:kafka-streams-internals-TableProcessorNode.adoc[TableProcessorNode]
 .. link:kafka-streams-internals-ProcessorGraphNode.adoc[ProcessorGraphNode]
+.. link:kafka-streams-internals-StatefulProcessorNode.adoc[StatefulProcessorNode]
 
 .. link:kafka-streams-internals-GroupedTableOperationRepartitionNode.adoc[GroupedTableOperationRepartitionNode]
 ... link:kafka-streams-internals-GroupedTableOperationRepartitionNodeBuilder.adoc[GroupedTableOperationRepartitionNodeBuilder]
@@ -237,6 +238,7 @@
 .. link:kafka-streams-internals-ConsumedInternal.adoc[ConsumedInternal -- Internal Accessors to Consumed Metadata]
 .. link:kafka-streams-internals-ProducedInternal.adoc[ProducedInternal -- Internal Accessors to Produced Metadata]
 .. link:kafka-streams-internals-QuickUnion.adoc[QuickUnion]
+.. link:kafka-streams-internals-TopicsInfo.adoc[TopicsInfo]
 
 . link:kafka-streams-NodeFactory.adoc[NodeFactory]
 .. link:kafka-streams-ProcessorNodeFactory.adoc[ProcessorNodeFactory]

kafka-streams-internals-InternalTopologyBuilder.adoc

@@ -487,24 +487,15 @@ void connectSourceStoreAndTopic(
   final String topic)
 ----
 
-`connectSourceStoreAndTopic` registers the `sourceStoreName` with the `topic` in <<storeToChangelogTopic, storeToChangelogTopic>>.
+`connectSourceStoreAndTopic` adds the given `sourceStoreName` with the `topic` to <<storeToChangelogTopic, storeToChangelogTopic>> internal registry.
 
-`connectSourceStoreAndTopic` reports a `TopologyException` when <<storeToChangelogTopic, storeToChangelogTopic>> has `sourceStoreName` already been registered.
+`connectSourceStoreAndTopic` reports a `TopologyException` when <<storeToChangelogTopic, storeToChangelogTopic>> has `sourceStoreName` already registered.
 
 ```
 Source store [sourceStoreName] is already added.
 ```
 
-[NOTE]
-====
-`connectSourceStoreAndTopic` is used when:
-
-* `InternalStreamsBuilder` is requested to link:kafka-streams-internals-InternalStreamsBuilder.adoc#table[create a KTable for a topic]
-
-* `InternalTopologyBuilder` is requested to <<addGlobalStore, add a global state store to a topology>>
-
-* *(deprecated)* `TopologyBuilder` is requested to `connectSourceStoreAndTopic`
-====
+NOTE: `connectSourceStoreAndTopic` is used when `InternalTopologyBuilder` is requested to <<addGlobalStore, add a global state store to a topology>> and <<adjust, adjust>>.
 
 === [[connectProcessorAndStateStore]] Connecting State Store with Processor Node -- `connectProcessorAndStateStore` Internal Method
 
@@ -1027,3 +1018,14 @@ InternalTopologyBuilder rewriteTopology(final StreamsConfig config)
 `rewriteTopology`...FIXME
 
 NOTE: `rewriteTopology` is used exclusively when `KafkaStreams` is <<kafka-streams-KafkaStreams.adoc#creating-instance, created>>.
+
+=== [[adjust]] `adjust` Internal Method
+
+[source, java]
+----
+void adjust(final StreamsConfig config)
+----
+
+`adjust`...FIXME
+
+NOTE: `adjust` is used exclusively when `InternalTopologyBuilder` is requested to <<rewriteTopology, rewriteTopology>>.

kafka-streams-internals-KStreamImpl.adoc

@@ -10,6 +10,9 @@
 [[SINK_NAME]]
 `KStreamImpl` uses *KSTREAM-SINK-* as the prefix for...FIXME
 
+[[TRANSFORMVALUES_NAME]]
+`KStreamImpl` uses *KSTREAM-TRANSFORMVALUES-* as the prefix for...FIXME
+
 [[internal-registries]]
 .KStreamImpl's Internal Properties (e.g. Registries, Counters and Flags)
 [cols="1,2",options="header",width="100%"]
@@ -327,3 +330,27 @@ KStream<K, V> filterNot(final Predicate<? super K, ? super V> predicate)
 NOTE: `filterNot` is part of the <<kafka-streams-KStream.adoc#filterNot, KStream Contract>> to...FIXME.
 
 `filterNot`...FIXME
+
+=== [[doTransformValues]] `doTransformValues` Internal Method
+
+[source, java]
+----
+KStream<K, VR> doTransformValues(
+  final ValueTransformerWithKeySupplier<
+    ? super K,
+    ? super V,
+    ? extends VR> valueTransformerWithKeySupplier,
+  final String... stateStoreNames)
+----
+
+`doTransformValues` requests the <<builder, InternalStreamsBuilder>> for a <<kafka-streams-internals-InternalStreamsBuilder.adoc#newProcessorName, new processor name>> with <<TRANSFORMVALUES_NAME, KSTREAM-TRANSFORMVALUES>> prefix.
+
+`doTransformValues` creates a new <<kafka-streams-internals-StatefulProcessorNode.adoc#, StatefulProcessorNode>> with the new processor name, the given `stateStoreNames` and the <<repartitionRequired, repartitionRequired>> flag.
+
+`doTransformValues` requests the `StatefulProcessorNode` to <<kafka-streams-internals-StreamsGraphNode.adoc#setValueChangingOperation, setValueChangingOperation>>.
+
+`doTransformValues` requests the <<builder, InternalStreamsBuilder>> to <<kafka-streams-internals-InternalStreamsBuilder.adoc#addGraphNode, add>> the `StatefulProcessorNode` (with the <<kafka-streams-AbstractStream.adoc#streamsGraphNode, StreamsGraphNode>> as the parent).
+
+In the end, `doTransformValues` creates a new <<creating-instance, KStreamImpl>> (with the new processor name, the <<sourceNodes, sourceNodes>>, the <<repartitionRequired, repartitionRequired>> flag, the `StatefulProcessorNode` itself and the <<builder, InternalStreamsBuilder>>).
+
+NOTE: `doTransformValues` is used when `KStreamImpl` is requested to <<transformValues, transformValues>>.

kafka-streams-internals-StatefulProcessorNode.adoc

@@ -0,0 +1,3 @@
+== [[StatefulProcessorNode]] StatefulProcessorNode
+
+`StatefulProcessorNode` is...FIXME

kafka-streams-internals-StreamsGraphNode.adoc

@@ -19,6 +19,10 @@ void writeToTopology(final InternalTopologyBuilder topologyBuilder)
 
 NOTE: `StreamsGraphNode` is a Java abstract class and cannot be <<creating-instance, created>> directly. It is created indirectly for the <<implementations, concrete StreamsGraphNodes>> and as the <<kafka-streams-internals-InternalStreamsBuilder.adoc#root, root node of an InternalStreamsBuilder>>.
 
+[[valueChangingOperation]]
+[[setValueChangingOperation]]
+`StreamsGraphNode` uses the `valueChangingOperation` flag for...FIXME
+
 [[implementations]]
 .StreamsGraphNodes (Direct Implementations)
 [cols="1,2",options="header",width="100%"]

kafka-streams-internals-TopicsInfo.adoc

@@ -0,0 +1,11 @@
+== [[TopicsInfo]] TopicsInfo
+
+[[creating-instance]]
+`TopicsInfo` is a simple "container" with the following:
+
+* [[sinkTopics]] Names of the sink topics
+* [[sourceTopics]] Names of the source topics
+* [[repartitionSourceTopics]] `Map<String, InternalTopicConfig>`
+* [[stateChangelogTopics]] `Map<String, InternalTopicConfig>`
+
+`TopicsInfo` is <<creating-instance, created>> exclusively when `InternalTopologyBuilder` is requested for the <<kafka-streams-internals-InternalTopologyBuilder.adoc#topicGroups, topic groups>> (when a node group has at least one source topic, incl. repartition or state changelog topics).