mjsax commented on code in PR #17198:
URL: https://github.com/apache/kafka/pull/17198#discussion_r1828630429
##########
streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/kstream/KStreamTest.scala:
##########
@@ -478,27 +479,51 @@ class KStreamTest extends TestDriver {
assertEquals("my-name", joinNode.name())
}
-// @nowarn
-// @Test
-// def testSettingNameOnTransform(): Unit = {
-// class TestTransformer extends Transformer[String, String,
KeyValue[String, String]] {
-// override def init(context: ProcessorContext): Unit = {}
-//
-// override def transform(key: String, value: String): KeyValue[String,
String] =
-// new KeyValue(s"$key-transformed", s"$value-transformed")
-//
-// override def close(): Unit = {}
-// }
-// val builder = new StreamsBuilder()
-// val sourceTopic = "source"
-// val sinkTopic = "sink"
-//
-// val stream = builder.stream[String, String](sourceTopic)
-// stream
-// .transform(() => new TestTransformer, Named.as("my-name"))
-// .to(sinkTopic)
-//
-// val transformNode =
builder.build().describe().subtopologies().asScala.head.nodes().asScala.toList(1)
-// assertEquals("my-name", transformNode.name())
-// }
+ @Test
+ def testSettingNameOnTransform(): Unit = {
+ val processorSupplier: ProcessorSupplier[String, String, String, String] =
+ new api.ProcessorSupplier[String, String, String, String] {
Review Comment:
Why do we need an explicit `ProcessorSupplier` for this test? And we use
`process(() => new TestProcessor)` similar to the old code?
##########
streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/kstream/KStreamTest.scala:
##########
@@ -478,27 +479,51 @@ class KStreamTest extends TestDriver {
assertEquals("my-name", joinNode.name())
}
-// @nowarn
-// @Test
-// def testSettingNameOnTransform(): Unit = {
-// class TestTransformer extends Transformer[String, String,
KeyValue[String, String]] {
-// override def init(context: ProcessorContext): Unit = {}
-//
-// override def transform(key: String, value: String): KeyValue[String,
String] =
-// new KeyValue(s"$key-transformed", s"$value-transformed")
-//
-// override def close(): Unit = {}
-// }
-// val builder = new StreamsBuilder()
-// val sourceTopic = "source"
-// val sinkTopic = "sink"
-//
-// val stream = builder.stream[String, String](sourceTopic)
-// stream
-// .transform(() => new TestTransformer, Named.as("my-name"))
-// .to(sinkTopic)
-//
-// val transformNode =
builder.build().describe().subtopologies().asScala.head.nodes().asScala.toList(1)
-// assertEquals("my-name", transformNode.name())
-// }
+ @Test
+ def testSettingNameOnTransform(): Unit = {
+ val processorSupplier: ProcessorSupplier[String, String, String, String] =
+ new api.ProcessorSupplier[String, String, String, String] {
+ private val storeName = "store-name"
+
+ override def stores: util.Set[StoreBuilder[_]] = {
+ val keyValueStoreBuilder = Stores.keyValueStoreBuilder(
+ Stores.persistentKeyValueStore(storeName),
+ Serdes.stringSerde,
+ Serdes.stringSerde
+ )
+ Collections.singleton(keyValueStoreBuilder)
+ }
+
+ override def get(): Processor[String, String, String, String] =
+ new api.Processor[String, String, String, String] {
+ private var context: api.ProcessorContext[String, String] = _
+ private var store: KeyValueStore[String, String] = _
+
+ override def init(context: api.ProcessorContext[String, String]):
Unit = {
+ this.context = context
+ store = context.getStateStore(storeName)
+ }
+
+ override def process(record: api.Record[String, String]): Unit = {
+ val key = record.key()
+ val value = record.value()
+ val processedKey = s"$key-processed"
+ val processedValue = s"$value-processed"
+ store.put(processedKey, processedValue)
+ context.forward(new api.Record(processedKey, processedValue,
record.timestamp()))
Review Comment:
What the `Processor` does is actually pretty meaningless for the test, so I
think the while implementation could actually be empty? For this case, we don't
even need `init()` method, as we don't need access to the `context` either.
##########
streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/kstream/KStreamTest.scala:
##########
@@ -478,27 +479,51 @@ class KStreamTest extends TestDriver {
assertEquals("my-name", joinNode.name())
}
-// @nowarn
-// @Test
-// def testSettingNameOnTransform(): Unit = {
-// class TestTransformer extends Transformer[String, String,
KeyValue[String, String]] {
-// override def init(context: ProcessorContext): Unit = {}
-//
-// override def transform(key: String, value: String): KeyValue[String,
String] =
-// new KeyValue(s"$key-transformed", s"$value-transformed")
-//
-// override def close(): Unit = {}
-// }
-// val builder = new StreamsBuilder()
-// val sourceTopic = "source"
-// val sinkTopic = "sink"
-//
-// val stream = builder.stream[String, String](sourceTopic)
-// stream
-// .transform(() => new TestTransformer, Named.as("my-name"))
-// .to(sinkTopic)
-//
-// val transformNode =
builder.build().describe().subtopologies().asScala.head.nodes().asScala.toList(1)
-// assertEquals("my-name", transformNode.name())
-// }
+ @Test
+ def testSettingNameOnTransform(): Unit = {
+ val processorSupplier: ProcessorSupplier[String, String, String, String] =
+ new api.ProcessorSupplier[String, String, String, String] {
+ private val storeName = "store-name"
Review Comment:
Why do we add a store?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
