vvcephei commented on a change in pull request #9253:
URL: https://github.com/apache/kafka/pull/9253#discussion_r567010831
##########
File path:
streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/Serdes.scala
##########
@@ -41,8 +43,19 @@ object Serdes {
implicit def Integer: Serde[Int] = JSerdes.Integer().asInstanceOf[Serde[Int]]
implicit def JavaInteger: Serde[java.lang.Integer] = JSerdes.Integer()
- implicit def timeWindowedSerde[T](implicit tSerde: Serde[T]):
WindowedSerdes.TimeWindowedSerde[T] =
- new WindowedSerdes.TimeWindowedSerde[T](tSerde)
+ implicit def timeWindowedSerde[T](implicit inner: Serde[T]):
Serde[Windowed[T]] =
+ new JSerdes.WrapperSerde[Windowed[T]](
+ new TimeWindowedSerializer[T](inner.serializer),
+ new TimeWindowedDeserializer[T](inner.deserializer) {
+ override def deserialize(topic: String, data: Array[Byte]):
Windowed[T] = {
Review comment:
Sorry if I'm missing the point here, @lct45 , but why do we need any
custom code at all in the scala wrapper? Shouldn't we just wrap the Java Serde,
which you've already set up to warn upon `configure` if there's no constructor
arg and no configured value?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
