KStreams Scala API: incorrect Javadocs and misleading parameter name
Reviewer: Matthias J. Sax Debasish Ghosh Guozhang Wang Bill Bejeck
@@ -261,30 +261,30 @@ class KStream[K, V](val inner: KStreamJ[K, V]) {
/**
* Transform each record of the input stream into zero or more records in the output stream (both key and value type
* can be altered arbitrarily).
- * A `Transformer` (provided by the given `TransformerSupplier`) is applied to each input record and
- * computes zero or more output records. In order to assign a state, the state must be created and registered
- * beforehand via stores added via `addStateStore` or `addGlobalStore` before they can be connected to the `Transformer`
+ * A `Transformer` is applied to each input record and computes zero or more output records. In order to assign a
+ * state, the state must be created and registered beforehand via stores added via `addStateStore` or `addGlobalStore`
+ * before they can be connected to the `Transformer`
*
- * @param transformerSupplier a instance of `TransformerSupplier` that generates a `Transformer`
+ * @param transformer the `Transformer` instance
* @param stateStoreNames the names of the state stores used by the processor
* @return a [[KStream]] that contains more or less records with new key and value (possibly of different type)
* @see `org.apache.kafka.streams.kstream.KStream#transform`
*/
- def transform[K1, V1](transformerSupplier: Transformer[K, V, (K1, V1)],
+ def transform[K1, V1](transformer: Transformer[K, V, (K1, V1)],
stateStoreNames: String*): KStream[K1, V1] = {
val transformerSupplierJ: TransformerSupplier[K, V, KeyValue[K1, V1]] = new TransformerSupplier[K, V, KeyValue[K1, V1]] {
override def get(): Transformer[K, V, KeyValue[K1, V1]] = {
new Transformer[K, V, KeyValue[K1, V1]] {
override def transform(key: K, value: V): KeyValue[K1, V1] = {
- transformerSupplier.transform(key, value) match {
+ transformer.transform(key, value) match {
case (k1, v1) => KeyValue.pair(k1, v1)
case _ => null
}
}
- override def init(context: ProcessorContext): Unit = transformerSupplier.init(context)
+ override def init(context: ProcessorContext): Unit = transformer.init(context)
- override def close(): Unit = transformerSupplier.close()
+ override def close(): Unit = transformer.close()
}
}
}
