StatefulProcessorNode

Author: jaceklaskowski

Diff for: kafka-streams-internals-GroupedStreamAggregateBuilder.adoc

@@ -17,10 +17,10 @@
 [source, java]
 ----
 <T> KTable<K, T> build(
-  final KStreamAggProcessorSupplier<K, ?, V, T> aggregateSupplier,
-  final String functionName,
-  final StoreBuilder storeBuilder,
-  final boolean isQueryable)
+  KStreamAggProcessorSupplier<K, ?, V, T> aggregateSupplier,
+  String functionName,
+  StoreBuilder storeBuilder,
+  boolean isQueryable)
 ----
 
 `build`...FIXME

Diff for: kafka-streams-internals-KStreamImpl.adoc

@@ -77,8 +77,8 @@ NOTE: `transformValues` is part of link:kafka-streams-KStream.adoc#transformValu
 [source, java]
 ----
 void process(
-  final ProcessorSupplier<? super K, ? super V> processorSupplier,
-  final String... stateStoreNames)
+  ProcessorSupplier<? super K, ? super V> processorSupplier,
+  String... stateStoreNames)
 ----
 
 NOTE: `process` is part of link:kafka-streams-KStream.adoc#process[KStream Contract] for...FIXME
@@ -90,8 +90,11 @@ NOTE: `process` is part of link:kafka-streams-KStream.adoc#process[KStream Contr
 [source, java]
 ----
 KStream<K1, V1> transform(
-  final TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier,
-  final String... stateStoreNames)
+  TransformerSupplier<
+    ? super K,
+    ? super V,
+    KeyValue<K1, V1>> transformerSupplier,
+  String... stateStoreNames)
 ----
 
 NOTE: `transform` is part of <<kafka-streams-KStream.adoc#transform, KStream API>> to transform a record into zero or more output records with state.
@@ -331,16 +334,53 @@ NOTE: `filterNot` is part of the <<kafka-streams-KStream.adoc#filterNot, KStream
 
 `filterNot`...FIXME
 
+=== [[flatTransform]] `flatTransform` Method
+
+[source, java]
+----
+KStream<K1, V1> flatTransform(
+  TransformerSupplier<
+    ? super K,
+    ? super V,
+    Iterable<KeyValue<K1, V1>>> transformerSupplier,
+  String... stateStoreNames)
+----
+
+NOTE: `flatTransform` is part of the <<kafka-streams-KStream.adoc#flatTransform, KStream Contract>> to...FIXME.
+
+`flatTransform`...FIXME
+
+=== [[flatTransformValues]] `flatTransformValues` Method
+
+[source, java]
+----
+KStream<K, VR> flatTransformValues(
+  ValueTransformerSupplier<
+    ? super V,
+    Iterable<VR>> valueTransformerSupplier,
+  String... stateStoreNames)
+KStream<K, VR> flatTransformValues(
+  ValueTransformerWithKeySupplier<
+    ? super K,
+    ? super V,
+    Iterable<VR>> valueTransformerSupplier,
+  String... stateStoreNames)
+----
+
+NOTE: `flatTransformValues` is part of the <<kafka-streams-KStream.adoc#flatTransformValues, KStream Contract>> to...FIXME.
+
+`flatTransformValues`...FIXME
+
 === [[doTransformValues]] `doTransformValues` Internal Method
 
 [source, java]
 ----
 KStream<K, VR> doTransformValues(
-  final ValueTransformerWithKeySupplier<
+  ValueTransformerWithKeySupplier<
     ? super K,
     ? super V,
     ? extends VR> valueTransformerWithKeySupplier,
-  final String... stateStoreNames)
+  String... stateStoreNames)
 ----
 
 `doTransformValues` requests the <<builder, InternalStreamsBuilder>> for a <<kafka-streams-internals-InternalStreamsBuilder.adoc#newProcessorName, new processor name>> with <<TRANSFORMVALUES_NAME, KSTREAM-TRANSFORMVALUES>> prefix.
@@ -353,7 +393,7 @@ KStream<K, VR> doTransformValues(
 
 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>>.
+NOTE: `doTransformValues` is used exclusively when `KStreamImpl` is requested to <<transformValues, transformValues>>.
 
 === [[internalSelectKey]] `internalSelectKey` Internal Method
 
@@ -366,3 +406,35 @@ ProcessorGraphNode<K, V> internalSelectKey(
 `internalSelectKey`...FIXME
 
 NOTE: `internalSelectKey` is used when `KStreamImpl` is requested to <<selectKey, selectKey>> and <<groupBy, groupBy>>.
+
+=== [[doFlatTransform]] `doFlatTransform` Internal Method
+
+[source, java]
+----
+KStream<K1, V1> doFlatTransform(
+  TransformerSupplier<
+    ? super K,
+    ? super V,
+    Iterable<KeyValue<K1, V1>>> transformerSupplier,
+  String... stateStoreNames)
+----
+
+`doFlatTransform`...FIXME
+
+NOTE: `doFlatTransform` is used when `KStreamImpl` is requested to <<transform, transform>> and <<flatTransform, flatTransform>>.
+
+=== [[doFlatTransformValues]] `doFlatTransformValues` Internal Method
+
+[source, java]
+----
+KStream<K, VR> doFlatTransformValues(
+  ValueTransformerWithKeySupplier<
+    ? super K,
+    ? super V,
+    Iterable<VR>> valueTransformerWithKeySupplier,
+  String... stateStoreNames)
+----
+
+`doFlatTransformValues`...FIXME
+
+NOTE: `doFlatTransformValues` is used exclusively when `KStreamImpl` is requested to <<flatTransformValues, flatTransformValues>>.

Diff for: kafka-streams-internals-StatefulProcessorNode.adoc

@@ -1,41 +1,48 @@
 == [[StatefulProcessorNode]] StatefulProcessorNode
 
-`StatefulProcessorNode` is a <<kafka-streams-internals-StreamsGraphNode.adoc#, graph node>> for stateful operators in <<kafka-streams-internals-KGroupedStreamImpl.adoc#, KGroupedStreamImpl>>, <<kafka-streams-internals-KGroupedTableImpl.adoc#, KGroupedTableImpl>>, <<kafka-streams-internals-KStreamImpl.adoc#, KStreamImpl>>, and <<kafka-streams-internals-KTableImpl.adoc#, KTableImpl>>.
+`StatefulProcessorNode` is a concrete <<kafka-streams-internals-StreamsGraphNode.adoc#, StreamsGraphNode>> (as a <<kafka-streams-internals-ProcessorGraphNode.adoc#, ProcessorGraphNode>>) that represents stateful operators of the following:
 
-[[storeBuilder]]
-`StatefulProcessorNode` extends the parent <<kafka-streams-internals-ProcessorGraphNode.adoc#, ProcessorGraphNode>> with the two state-related parameters, i.e. <<storeNames, state store names>> and <<materializedKTableStoreBuilder, StoreBuilder>>.
+* <<kafka-streams-internals-KGroupedStreamImpl.adoc#, KGroupedStreamImpl>>
+
+* <<kafka-streams-internals-KGroupedTableImpl.adoc#, KGroupedTableImpl>>
 
-When requested to <<writeToTopology, writeToTopology>>, `StatefulProcessorNode` simply requests the given <<kafka-streams-internals-InternalTopologyBuilder.adoc#, InternalTopologyBuilder>> to <<kafka-streams-internals-InternalTopologyBuilder.adoc#addProcessor, add a processor>> (as a <<kafka-streams-internals-ProcessorGraphNode.adoc#, ProcessorGraphNode>>) and to associate the state stores (by the <<storeNames, names>> or the <<materializedKTableStoreBuilder, StoreBuilder>>).
+* <<kafka-streams-internals-KStreamImpl.adoc#, KStreamImpl>>
+
+* <<kafka-streams-internals-KTableImpl.adoc#, KTableImpl>>
 
 `StatefulProcessorNode` is <<creating-instance, created>> when:
 
-* `GroupedStreamAggregateBuilder` is requested to <<kafka-streams-internals-GroupedStreamAggregateBuilder.adoc#build, build>> (for <<kafka-streams-internals-KStreamImpl.adoc#groupBy, KStream.groupBy>> and <<kafka-streams-internals-KStreamImpl.adoc#groupByKey, KStream.groupByKey>> streaming operators, incl. `windowedBy` operator with a <<kafka-streams-internals-KGroupedStreamImpl.adoc#windowedBy-Windows, Windows>> or a <<kafka-streams-internals-KGroupedStreamImpl.adoc#windowedBy-SessionWindows, SessionWindows>>)
+* `KStreamImpl` is requested to <<kafka-streams-internals-KStreamImpl.adoc#transform, KStreamImpl.transform>> and <<kafka-streams-internals-KStreamImpl.adoc#flatTransform, KStreamImpl.flatTransform>>, <<kafka-streams-internals-KStreamImpl.adoc#transformValues, KStreamImpl.transformValues>>, <<kafka-streams-internals-KStreamImpl.adoc#flatTransformValues, KStreamImpl.flatTransformValues>>, <<kafka-streams-internals-KStreamImpl.adoc#process, KStreamImpl.process>>
 
-* `KGroupedTableImpl` is requested to <<kafka-streams-internals-KGroupedTableImpl.adoc#doAggregate, doAggregate>> (for <<kafka-streams-internals-KGroupedTableImpl.adoc#reduce, reduce>>, <<kafka-streams-internals-KGroupedTableImpl.adoc#count, count>> and <<kafka-streams-internals-KGroupedTableImpl.adoc#aggregate, aggregate>> operators)
+* `GroupedStreamAggregateBuilder` is requested to <<kafka-streams-internals-GroupedStreamAggregateBuilder.adoc#build, build>> (for <<kafka-streams-internals-KStreamImpl.adoc#groupBy, KStream.groupBy>> and <<kafka-streams-internals-KStreamImpl.adoc#groupByKey, KStream.groupByKey>> streaming operators, incl. `windowedBy` operator with a <<kafka-streams-internals-KGroupedStreamImpl.adoc#windowedBy-Windows, Windows>> or a <<kafka-streams-internals-KGroupedStreamImpl.adoc#windowedBy-SessionWindows, SessionWindows>>)
 
-* `KStreamImpl` is requested to <<kafka-streams-internals-KStreamImpl.adoc#transform, transform>>, <<kafka-streams-internals-KStreamImpl.adoc#doTransformValues, doTransformValues>> (for <<transformValues, transformValues>> operator), and <<kafka-streams-internals-KStreamImpl.adoc#process, process>>
+* `KGroupedTableImpl` is requested to <<kafka-streams-internals-KGroupedTableImpl.adoc#doAggregate, doAggregate>> (for <<kafka-streams-internals-KGroupedTableImpl.adoc#aggregate, KGroupedTableImpl.aggregate>>, <<kafka-streams-internals-KGroupedTableImpl.adoc#count, KGroupedTableImpl.count>>, and <<kafka-streams-internals-KGroupedTableImpl.adoc#reduce, KGroupedTableImpl.reduce>> operators)
 
 * `KTableImpl` is requested to <<kafka-streams-internals-KTableImpl.adoc#suppress, suppress>>
 
-[[creating-instance]]
+[[storeBuilder]]
+`StatefulProcessorNode` extends the parent <<kafka-streams-internals-ProcessorGraphNode.adoc#, ProcessorGraphNode>> with the two state-related parameters - the <<storeNames, state store names>> and a <<kafka-streams-StoreBuilder.adoc#, StoreBuilder>>.
+
+=== [[creating-instance]] Creating StatefulProcessorNode Instance
+
 `StatefulProcessorNode` takes the following to be created:
 
 * [[nodeName]] Node name
 * [[processorParameters]] `ProcessorParameters<K, V>`
 * [[materializedKTableStoreBuilder]][[storeNames]] Names of the state stores or a <<kafka-streams-StoreBuilder.adoc#, StoreBuilder>>
-* [[repartitionRequired]] `repartitionRequired` flag
 
 === [[writeToTopology]] `writeToTopology` Method
 
 [source, java]
 ----
-void writeToTopology(final InternalTopologyBuilder topologyBuilder)
+void writeToTopology(
+  InternalTopologyBuilder topologyBuilder)
 ----
 
 NOTE: `writeToTopology` is part of the <<kafka-streams-internals-StreamsGraphNode.adoc#writeToTopology, StreamsGraphNode Contract>> to...FIXME.
 
-`writeToTopology` requests the given `InternalTopologyBuilder` to <<kafka-streams-internals-InternalTopologyBuilder.adoc#addProcessor, add a processor>> with the name and the <<kafka-streams-ProcessorSupplier.adoc#, ProcessorSupplier>> as defined by the <<processorParameters, ProcessorParameters>>.
+`writeToTopology` requests the given <<kafka-streams-internals-InternalTopologyBuilder.adoc#, InternalTopologyBuilder>> to <<kafka-streams-internals-InternalTopologyBuilder.adoc#addProcessor, add a processor>> (with the name and the <<kafka-streams-ProcessorSupplier.adoc#, ProcessorSupplier>> as defined by the <<processorParameters, ProcessorParameters>>).
 
-With <<storeNames, state store name>> given, `writeToTopology` requests the given `InternalTopologyBuilder` to <<kafka-streams-internals-InternalTopologyBuilder.adoc#connectProcessorAndStateStores, connect them with the processor>>.
+With <<storeNames, state store names>> given, `writeToTopology` requests the given `InternalTopologyBuilder` to <<kafka-streams-internals-InternalTopologyBuilder.adoc#connectProcessorAndStateStores, connect them with the processor>>.
 
-With a <<storeBuilder, StoreBuilder>> given, `writeToTopology` requests the given `InternalTopologyBuilder` to <<kafka-streams-internals-InternalTopologyBuilder.adoc#addStateStore, addStateStore>> with the `StoreBuilder` and the processor.
+With a <<storeBuilder, StoreBuilder>> given, `writeToTopology` requests the given `InternalTopologyBuilder` to <<kafka-streams-internals-InternalTopologyBuilder.adoc#addStateStore, add a state store>> (with the `StoreBuilder` and the processor).