Transformer -- Record-by-Record Stateful Transformation

Author: jaceklaskowski

SUMMARY.adoc

@@ -36,7 +36,8 @@
 .. link:kafka-streams-StateRestoreCallback.adoc[StateRestoreCallback]
 .. link:kafka-streams-To.adoc[To]
 
-. link:kafka-streams-Transformer.adoc[Transformer]
+. link:kafka-streams-Transformer.adoc[Transformer -- Record-by-Record Stateful Transformation]
+.. link:kafka-streams-TransformerSupplier.adoc[TransformerSupplier]
 
 . link:kafka-streams-ValueTransformer.adoc[ValueTransformer -- Stateful Record-by-Record Value Transformation]
 .. link:kafka-streams-ValueTransformerSupplier.adoc[ValueTransformerSupplier -- ValueTransformers Object Factory]
@@ -136,6 +137,7 @@
 . link:kafka-streams-Processor.adoc[Processor Contract -- Stream Processing Nodes]
 .. link:kafka-streams-AbstractProcessor.adoc[AbstractProcessor -- Base for Stream Processors]
 .. link:kafka-streams-KStreamSessionWindowAggregateProcessor.adoc[KStreamSessionWindowAggregateProcessor]
+.. link:kafka-streams-KStreamTransformProcessor.adoc[KStreamTransformProcessor]
 .. link:kafka-streams-KStreamTransformValuesProcessor.adoc[KStreamTransformValuesProcessor]
 
 . link:kafka-streams-ProcessorNodePunctuator.adoc[ProcessorNodePunctuator]

kafka-streams-KStreamTransformProcessor.adoc

@@ -0,0 +1,36 @@
+== [[KStreamTransformProcessor]] KStreamTransformProcessor
+
+`KStreamTransformProcessor` is...FIXME
+
+=== [[close]] Closing Processor -- `close` Method
+
+[source, java]
+----
+void close()
+----
+
+NOTE: `close` is part of the <<kafka-streams-Processor.adoc#close, Processor Contract>> to close the <<kafka-streams-Processor.adoc#, processor>>.
+
+`close`...FIXME
+
+=== [[init]] Initializing Processor -- `init` Method
+
+[source, java]
+----
+void init(ProcessorContext context)
+----
+
+NOTE: `init` is part of the <<kafka-streams-Processor.adoc#init, Processor Contract>> to initialize the <<kafka-streams-Processor.adoc#, processor>>.
+
+`init`...FIXME
+
+=== [[process]] Processing Record -- `process` Method
+
+[source, java]
+----
+void process(K1 key, V1 value)
+----
+
+NOTE: `process` is part of the <<kafka-streams-Processor.adoc#process, Processor Contract>> to process a single record.
+
+`process`...FIXME

kafka-streams-Processor.adoc

@@ -5,36 +5,41 @@
 `Processor` is an integral part of link:kafka-streams-ProcessorNode.adoc#processor[ProcessorNode].
 
 [[contract]]
+.Processor Contract
+[cols="1m,2",options="header",width="100%"]
+|===
+| Method
+| Description
+
+| close
+a| [[close]]
+
 [source, java]
 ----
-package org.apache.kafka.streams.processor;
-
-interface Processor<K, V> {
-  void close();
-  void init(ProcessorContext context);
-  void process(K key, V value);
-}
+void close()
 ----
 
-NOTE: `Processor` is an `Evolving` contract which means that compatibility may be broken at a minor release.
+Closes the processor
 
-.Processor Contract
-[cols="1,2",options="header",width="100%"]
-|===
-| Method
-| Description
+| init
+a| [[init]]
 
-| `close`
-| [[close]]
+[source, java]
+----
+void init(ProcessorContext context)
+----
 
-| `init`
-| [[init]] Initializing a stream processor
+Initializes the processor
 
-| `process`
-| [[process]] Processing a single record
+| process
+a| [[process]]
+
+[source, java]
+----
+void process(K key, V value)
+----
 
-| `punctuate`
-| [[punctuate]] *DEPRECATED*
+Processes a single record
 |===
 
 The lifecycle of a `Processor` starts at <<init, init>> when....FIXME

kafka-streams-Transformer.adoc

@@ -1,3 +1,59 @@
-== [[Transformer]] Transformer
+== [[Transformer]] Transformer -- Record-by-Record Stateful Transformation
 
-`Transformer` is...FIXME
+`Transformer` is the <<contract, contract>> that allows for a <<transform, stateful transformation>> of every record in the input stream into zero or more output records (both key and value type can be altered arbitrarily).
+
+In order to use a state, a state store has to be registered beforehand (using `addStateStore` or `addGlobalStore`) before they can be connected to the `Transformer`.
+
+You can access a state store using the <<kafka-streams-ProcessorContext.adoc#, ProcessorContext>> that is given when a transformer is <<init, initialized>>.
+
+[source, scala]
+----
+// FIXME: Demo
+----
+
+`Transformer` is created (_supplied_) using a corresponding <<kafka-streams-TransformerSupplier.adoc#, TransformerSupplier>>.
+
+[[contract]]
+[cols="1m,2",options="header",width="100%"]
+|===
+| Property
+| Description
+
+| close
+a| [[close]]
+
+[source, java]
+----
+void close()
+----
+
+Closes the transformer
+
+Used exclusively when `KStreamTransformProcessor` is requested to <<kafka-streams-KStreamTransformProcessor.adoc#close, close>>
+
+| init
+a| [[init]]
+
+[source, java]
+----
+void init(final ProcessorContext context)
+----
+
+Initializes the transformer in the given <<kafka-streams-ProcessorContext.adoc#, ProcessorContext>>
+
+Used exclusively when `KStreamTransformProcessor` is requested to <<kafka-streams-KStreamTransformProcessor.adoc#init, initialize itself>>
+
+| transform
+a| [[transform]]
+
+[source, java]
+----
+R transform(final K key, final V value)
+----
+
+Transforms a single record (into a `KeyValue` object)
+
+Used exclusively when `KStreamTransformProcessor` is requested to <<kafka-streams-KStreamTransformProcessor.adoc#process, process a record>>
+|===
+
+`Transformer` is used to create a <<kafka-streams-KStreamTransformProcessor.adoc#, KStreamTransformProcessor>>.

kafka-streams-TransformerSupplier.adoc

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