KAFKA-7396: Materialized, Serialized, Joined, Consumed and Produced with implicit Serdes (#5551)

We want to make sure that we always have a serde for all Materialized, Serialized, Joined, Consumed and Produced.
For that we can make use of the implicit parameters in Scala.

KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-365%3A+Materialized%2C+Serialized%2C+Joined%2C+Consumed+and+Produced+with+implicit+Serde

Reviewers: John Roesler <[email protected]>, Matthias J. Sax <[email protected]>, Bill Bejeck <[email protected]>, Guozhang Wang <[email protected]>, Ted Yu <[email protected]>

Author: joan38

streams/src/main/java/org/apache/kafka/streams/kstream/Produced.java

@@ -71,7 +71,8 @@ public static <K, V> Produced<K, V> with(final Serde<K> keySerde,
      * @param valueSerde    Serde to use for serializing the value
      * @param partitioner   the function used to determine how records are distributed among partitions of the topic,
      *                      if not specified and {@code keySerde} provides a {@link WindowedSerializer} for the key
-     *                      {@link WindowedStreamPartitioner} will be used&mdash;otherwise {@link DefaultPartitioner} wil be used
+     *                      {@link WindowedStreamPartitioner} will be used&mdash;otherwise {@link DefaultPartitioner}
+     *                      will be used
      * @param <K>           key type
      * @param <V>           value type
      * @return  A new {@link Produced} instance configured with keySerde, valueSerde, and partitioner

streams/src/main/java/org/apache/kafka/streams/kstream/internals/ProducedInternal.java

@@ -21,7 +21,7 @@
 import org.apache.kafka.streams.processor.StreamPartitioner;
 
 public class ProducedInternal<K, V> extends Produced<K, V> {
-    ProducedInternal(final Produced<K, V> produced) {
+    public ProducedInternal(final Produced<K, V> produced) {
         super(produced);
     }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/SerializedInternal.java

@@ -19,17 +19,16 @@
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.streams.kstream.Serialized;
 
-class SerializedInternal<K, V> extends Serialized<K, V> {
-    SerializedInternal(final Serialized<K, V> serialized) {
+public class SerializedInternal<K, V> extends Serialized<K, V> {
+    public SerializedInternal(final Serialized<K, V> serialized) {
         super(serialized);
     }
 
-    Serde<K> keySerde() {
+    public Serde<K> keySerde() {
         return keySerde;
     }
 
-    Serde<V> valueSerde() {
+    public Serde<V> valueSerde() {
         return valueSerde;
     }
-
 }

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/ImplicitConversions.scala

@@ -25,14 +25,13 @@ import org.apache.kafka.streams.kstream.{
   KStream => KStreamJ,
   KTable => KTableJ,
   SessionWindowedKStream => SessionWindowedKStreamJ,
-  TimeWindowedKStream => TimeWindowedKStreamJ,
-  _
+  TimeWindowedKStream => TimeWindowedKStreamJ
 }
 import org.apache.kafka.streams.scala.kstream._
 import org.apache.kafka.streams.KeyValue
 import org.apache.kafka.common.serialization.Serde
-import scala.language.implicitConversions
 
+import scala.language.implicitConversions
 import org.apache.kafka.streams.processor.StateStore
 
 /**
@@ -65,20 +64,20 @@ object ImplicitConversions {
   // and these implicits will convert them to `Serialized`, `Produced` or `Consumed`
 
   implicit def serializedFromSerde[K, V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Serialized[K, V] =
-    Serialized.`with`(keySerde, valueSerde)
+    Serialized.`with`[K, V]
 
   implicit def consumedFromSerde[K, V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Consumed[K, V] =
-    Consumed.`with`(keySerde, valueSerde)
+    Consumed.`with`[K, V]
 
   implicit def producedFromSerde[K, V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Produced[K, V] =
-    Produced.`with`(keySerde, valueSerde)
+    Produced.`with`[K, V]
 
   implicit def materializedFromSerde[K, V, S <: StateStore](implicit keySerde: Serde[K],
                                                             valueSerde: Serde[V]): Materialized[K, V, S] =
-    Materialized.`with`[K, V, S](keySerde, valueSerde)
+    Materialized.`with`[K, V, S]
 
   implicit def joinedFromKeyValueOtherSerde[K, V, VO](implicit keySerde: Serde[K],
                                                       valueSerde: Serde[V],
                                                       otherValueSerde: Serde[VO]): Joined[K, V, VO] =
-    Joined.`with`(keySerde, valueSerde, otherValueSerde)
+    Joined.`with`[K, V, VO]
 }

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/StreamsBuilder.scala

@@ -21,7 +21,7 @@ package org.apache.kafka.streams.scala
 
 import java.util.regex.Pattern
 
-import org.apache.kafka.streams.kstream.{Consumed, GlobalKTable, Materialized}
+import org.apache.kafka.streams.kstream.GlobalKTable
 import org.apache.kafka.streams.processor.{ProcessorSupplier, StateStore}
 import org.apache.kafka.streams.state.StoreBuilder
 import org.apache.kafka.streams.{Topology, StreamsBuilder => StreamsBuilderJ}

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/Consumed.scala

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala.kstream
+
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.streams.kstream.{Consumed => ConsumedJ}
+import org.apache.kafka.streams.Topology
+import org.apache.kafka.streams.processor.TimestampExtractor
+
+object Consumed {
+
+  /**
+   * Create an instance of [[Consumed]] with the supplied arguments. `null` values are acceptable.
+   *
+   * @tparam K                 key type
+   * @tparam V                 value type
+   * @param timestampExtractor the timestamp extractor to used. If `null` the default timestamp extractor from
+   *                           config will be used
+   * @param resetPolicy        the offset reset policy to be used. If `null` the default reset policy from config
+   *                           will be used
+   * @param keySerde           the key serde to use.
+   * @param valueSerde         the value serde to use.
+   * @return a new instance of [[Consumed]]
+   */
+  def `with`[K, V](
+    timestampExtractor: TimestampExtractor,
+    resetPolicy: Topology.AutoOffsetReset
+  )(implicit keySerde: Serde[K], valueSerde: Serde[V]): ConsumedJ[K, V] =
+    ConsumedJ.`with`(keySerde, valueSerde, timestampExtractor, resetPolicy)
+
+  /**
+   * Create an instance of [[Consumed]] with key and value [[Serde]]s.
+   *
+   * @tparam K         key type
+   * @tparam V         value type
+   * @return a new instance of [[Consumed]]
+   */
+  def `with`[K, V](implicit keySerde: Serde[K], valueSerde: Serde[V]): ConsumedJ[K, V] =
+    ConsumedJ.`with`(keySerde, valueSerde)
+
+  /**
+   * Create an instance of [[Consumed]] with a [[TimestampExtractor]].
+   *
+   * @param timestampExtractor the timestamp extractor to used. If `null` the default timestamp extractor from
+   *                           config will be used
+   * @tparam K                 key type
+   * @tparam V                 value type
+   * @return a new instance of [[Consumed]]
+   */
+  def `with`[K, V](timestampExtractor: TimestampExtractor)(implicit keySerde: Serde[K],
+                                                           valueSerde: Serde[V]): ConsumedJ[K, V] =
+    ConsumedJ.`with`(timestampExtractor).withKeySerde(keySerde).withValueSerde(valueSerde)
+
+  /**
+   * Create an instance of [[Consumed]] with a [[Topology.AutoOffsetReset]].
+   *
+   * @tparam K          key type
+   * @tparam V          value type
+   * @param resetPolicy the offset reset policy to be used. If `null` the default reset policy from config will be used
+   * @return a new instance of [[Consumed]]
+   */
+  def `with`[K, V](resetPolicy: Topology.AutoOffsetReset)(implicit keySerde: Serde[K],
+                                                          valueSerde: Serde[V]): ConsumedJ[K, V] =
+    ConsumedJ.`with`(resetPolicy).withKeySerde(keySerde).withValueSerde(valueSerde)
+}

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/Joined.scala

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala.kstream
+
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.streams.kstream.{Joined => JoinedJ}
+
+object Joined {
+
+  /**
+   * Create an instance of [[org.apache.kafka.streams.kstream.Joined]] with key, value, and otherValue [[Serde]]
+   * instances.
+   * `null` values are accepted and will be replaced by the default serdes as defined in config.
+   *
+   * @tparam K              key type
+   * @tparam V              value type
+   * @tparam VO             other value type
+   * @param keySerde        the key serde to use.
+   * @param valueSerde      the value serde to use.
+   * @param otherValueSerde the otherValue serde to use. If `null` the default value serde from config will be used
+   * @return new [[org.apache.kafka.streams.kstream.Joined]] instance with the provided serdes
+   */
+  def `with`[K, V, VO](implicit keySerde: Serde[K],
+                       valueSerde: Serde[V],
+                       otherValueSerde: Serde[VO]): JoinedJ[K, V, VO] =
+    JoinedJ.`with`(keySerde, valueSerde, otherValueSerde)
+
+}

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedStream.scala

@@ -27,8 +27,8 @@ import org.apache.kafka.streams.scala.FunctionConversions._
 /**
  * Wraps the Java class KGroupedStream and delegates method calls to the underlying Java object.
  *
- * @param [K] Type of keys
- * @param [V] Type of values
+ * @tparam K Type of keys
+ * @tparam V Type of values
  * @param inner The underlying Java abstraction for KGroupedStream
  *
  * @see `org.apache.kafka.streams.kstream.KGroupedStream`

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedTable.scala

@@ -27,8 +27,8 @@ import org.apache.kafka.streams.scala.FunctionConversions._
 /**
  * Wraps the Java class KGroupedTable and delegates method calls to the underlying Java object.
  *
- * @param [K] Type of keys
- * @param [V] Type of values
+ * @tparam K Type of keys
+ * @tparam V Type of values
  * @param inner The underlying Java abstraction for KGroupedTable
  *
  * @see `org.apache.kafka.streams.kstream.KGroupedTable`

streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/Materialized.scala

@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala.kstream
+
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.streams.kstream.{Materialized => MaterializedJ}
+import org.apache.kafka.streams.processor.StateStore
+import org.apache.kafka.streams.scala.{ByteArrayKeyValueStore, ByteArraySessionStore, ByteArrayWindowStore}
+import org.apache.kafka.streams.state.{KeyValueBytesStoreSupplier, SessionBytesStoreSupplier, WindowBytesStoreSupplier}
+
+object Materialized {
+
+  /**
+   * Materialize a [[StateStore]] with the provided key and value [[Serde]]s.
+   * An internal name will be used for the store.
+   *
+   * @tparam K         key type
+   * @tparam V         value type
+   * @tparam S         store type
+   * @param keySerde   the key [[Serde]] to use.
+   * @param valueSerde the value [[Serde]] to use.
+   * @return a new [[Materialized]] instance with the given key and value serdes
+   */
+  def `with`[K, V, S <: StateStore](implicit keySerde: Serde[K], valueSerde: Serde[V]): MaterializedJ[K, V, S] =
+    MaterializedJ.`with`(keySerde, valueSerde)
+
+  /**
+   * Materialize a [[StateStore]] with the given name.
+   *
+   * @tparam K         key type of the store
+   * @tparam V         value type of the store
+   * @tparam S         type of the [[StateStore]]
+   * @param storeName  the name of the underlying [[org.apache.kafka.streams.scala.kstream.KTable]] state store;
+   *                   valid characters are ASCII alphanumerics, '.', '_' and '-'.
+   * @param keySerde   the key serde to use.
+   * @param valueSerde the value serde to use.
+   * @return a new [[Materialized]] instance with the given storeName
+   */
+  def as[K, V, S <: StateStore](storeName: String)(implicit keySerde: Serde[K],
+                                                   valueSerde: Serde[V]): MaterializedJ[K, V, S] =
+    MaterializedJ.as(storeName).withKeySerde(keySerde).withValueSerde(valueSerde)
+
+  /**
+   * Materialize a [[org.apache.kafka.streams.state.WindowStore]] using the provided [[WindowBytesStoreSupplier]].
+   *
+   * Important: Custom subclasses are allowed here, but they should respect the retention contract:
+   * Window stores are required to retain windows at least as long as (window size + window grace period).
+   * Stores constructed via [[org.apache.kafka.streams.state.Stores]] already satisfy this contract.
+   *
+   * @tparam K         key type of the store
+   * @tparam V         value type of the store
+   * @param supplier   the [[WindowBytesStoreSupplier]] used to materialize the store
+   * @param keySerde   the key serde to use.
+   * @param valueSerde the value serde to use.
+   * @return a new [[Materialized]] instance with the given supplier
+   */
+  def as[K, V](supplier: WindowBytesStoreSupplier)(implicit keySerde: Serde[K],
+                                                   valueSerde: Serde[V]): MaterializedJ[K, V, ByteArrayWindowStore] =
+    MaterializedJ.as(supplier).withKeySerde(keySerde).withValueSerde(valueSerde)
+
+  /**
+   * Materialize a [[org.apache.kafka.streams.state.SessionStore]] using the provided [[SessionBytesStoreSupplier]].
+   *
+   * Important: Custom subclasses are allowed here, but they should respect the retention contract:
+   * Session stores are required to retain windows at least as long as (session inactivity gap + session grace period).
+   * Stores constructed via [[org.apache.kafka.streams.state.Stores]] already satisfy this contract.
+   *
+   * @tparam K         key type of the store
+   * @tparam V         value type of the store
+   * @param supplier   the [[SessionBytesStoreSupplier]] used to materialize the store
+   * @param keySerde   the key serde to use.
+   * @param valueSerde the value serde to use.
+   * @return a new [[Materialized]] instance with the given supplier
+   */
+  def as[K, V](supplier: SessionBytesStoreSupplier)(implicit keySerde: Serde[K],
+                                                    valueSerde: Serde[V]): MaterializedJ[K, V, ByteArraySessionStore] =
+    MaterializedJ.as(supplier).withKeySerde(keySerde).withValueSerde(valueSerde)
+
+  /**
+   * Materialize a [[org.apache.kafka.streams.state.KeyValueStore]] using the provided [[KeyValueBytesStoreSupplier]].
+   *
+   * @tparam K         key type of the store
+   * @tparam V         value type of the store
+   * @param supplier   the [[KeyValueBytesStoreSupplier]] used to materialize the store
+   * @param keySerde   the key serde to use.
+   * @param valueSerde the value serde to use.
+   * @return a new [[Materialized]] instance with the given supplier
+   */
+  def as[K, V](
+    supplier: KeyValueBytesStoreSupplier
+  )(implicit keySerde: Serde[K], valueSerde: Serde[V]): MaterializedJ[K, V, ByteArrayKeyValueStore] =
+    MaterializedJ.as(supplier).withKeySerde(keySerde).withValueSerde(valueSerde)
+}