You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2025/03/31 12:15:42 UTC
(kafka) branch trunk updated: KAFKA-16260: Deprecate window.size.ms and window.inner.class.serde in StreamsConfig (#18297)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new c5f8ae0424a KAFKA-16260: Deprecate window.size.ms and window.inner.class.serde in StreamsConfig (#18297)
c5f8ae0424a is described below
commit c5f8ae0424ad189cca87126e55a4e17c676578d2
Author: PoAn Yang <pa...@apache.org>
AuthorDate: Mon Mar 31 20:15:37 2025 +0800
KAFKA-16260: Deprecate window.size.ms and window.inner.class.serde in StreamsConfig (#18297)
The `window.size.ms` and `window.inner.class.serde` are not a true
KafkaStreams config, and are ignored when set from a KStreams
application. Both belong on the client.
Reviewers: Lucas Brutschy <lb...@confluent.io>
Signed-off-by: PoAn Yang <pa...@apache.org>
---
docs/upgrade.html | 8 ++
.../KStreamAggregationIntegrationTest.java | 20 ++--
.../SelfJoinUpgradeIntegrationTest.java | 3 +-
.../SlidingWindowedKStreamIntegrationTest.java | 9 +-
.../TimeWindowedKStreamIntegrationTest.java | 9 +-
.../org/apache/kafka/streams/StreamsConfig.java | 23 ++++-
.../kstream/SessionWindowedDeserializer.java | 49 ++++++---
.../streams/kstream/SessionWindowedSerializer.java | 48 ++++++---
.../streams/kstream/TimeWindowedDeserializer.java | 115 +++++++++++++--------
.../streams/kstream/TimeWindowedSerializer.java | 48 ++++++---
.../kstream/SessionWindowedDeserializerTest.java | 37 ++++++-
.../kstream/SessionWindowedSerializerTest.java | 37 ++++++-
.../kstream/TimeWindowedDeserializerTest.java | 71 +++++++++++--
.../kstream/TimeWindowedSerializerTest.java | 36 ++++++-
14 files changed, 394 insertions(+), 119 deletions(-)
diff --git a/docs/upgrade.html b/docs/upgrade.html
index 88be13c531e..3ffa47f0b02 100644
--- a/docs/upgrade.html
+++ b/docs/upgrade.html
@@ -36,6 +36,14 @@
</li>
</ul>
</li>
+ <li><b>Stream</b>
+ <ul>
+ <li>
+ The <code>window.size.ms</code> and <code>window.inner.serde.class</code> in <code>StreamsConfig</code> are deprecated.
+ Use the corresponding string constants defined in <code>TimeWindowedSerializer</code>, <code>TimeWindowedDeserializer</code>, <code>SessionWindowedSerializer</code> and <code>SessionWindowedDeserializer</code> instead.
+ </li>
+ </ul>
+ </li>
</ul>
<h4><a id="upgrade_4_0_0" href="#upgrade_4_0_0">Upgrading to 4.0.0</a></h4>
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
index bf0d54bc5c0..7037ce1368d 100644
--- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
@@ -1065,9 +1065,12 @@ public class KStreamAggregationIntegrationTest {
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
- consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, 500L);
- if (keyDeserializer instanceof TimeWindowedDeserializer || keyDeserializer instanceof SessionWindowedDeserializer) {
- consumerProperties.setProperty(StreamsConfig.WINDOWED_INNER_CLASS_SERDE,
+ consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, 500L);
+ if (keyDeserializer instanceof TimeWindowedDeserializer) {
+ consumerProperties.setProperty(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
+ Serdes.serdeFrom(innerClass).getClass().getName());
+ } else if (keyDeserializer instanceof SessionWindowedDeserializer) {
+ consumerProperties.setProperty(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
Serdes.serdeFrom(innerClass).getClass().getName());
}
return IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(
@@ -1089,9 +1092,12 @@ public class KStreamAggregationIntegrationTest {
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
- consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, 500L);
- if (keyDeserializer instanceof TimeWindowedDeserializer || keyDeserializer instanceof SessionWindowedDeserializer) {
- consumerProperties.setProperty(StreamsConfig.WINDOWED_INNER_CLASS_SERDE,
+ consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, 500L);
+ if (keyDeserializer instanceof TimeWindowedDeserializer) {
+ consumerProperties.setProperty(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
+ Serdes.serdeFrom(innerClass).getClass().getName());
+ } else if (keyDeserializer instanceof SessionWindowedDeserializer) {
+ consumerProperties.setProperty(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
Serdes.serdeFrom(innerClass).getClass().getName());
}
return IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(
@@ -1123,7 +1129,7 @@ public class KStreamAggregationIntegrationTest {
"--property", "key.deserializer=" + keyDeserializer.getClass().getName(),
"--property", "value.deserializer=" + valueDeserializer.getClass().getName(),
"--property", "key.separator=" + keySeparator,
- "--property", "key.deserializer." + StreamsConfig.WINDOWED_INNER_CLASS_SERDE + "=" + Serdes.serdeFrom(innerClass).getClass().getName(),
+ "--property", "key.deserializer." + TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS + "=" + Serdes.serdeFrom(innerClass).getClass().getName(),
"--property", "key.deserializer.window.size.ms=500",
};
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SelfJoinUpgradeIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SelfJoinUpgradeIntegrationTest.java
index d97d85a6af3..c9b498d4054 100644
--- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SelfJoinUpgradeIntegrationTest.java
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SelfJoinUpgradeIntegrationTest.java
@@ -32,6 +32,7 @@ import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.JoinWindows;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.streams.kstream.TimeWindowedDeserializer;
import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.test.TestUtils;
@@ -262,7 +263,7 @@ public class SelfJoinUpgradeIntegrationTest {
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
- consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, 500L);
+ consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, 500L);
final List<KeyValueTimestamp<K, V>> actual =
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SlidingWindowedKStreamIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SlidingWindowedKStreamIntegrationTest.java
index 5f972f7c4f4..2954fa9806d 100644
--- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SlidingWindowedKStreamIntegrationTest.java
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SlidingWindowedKStreamIntegrationTest.java
@@ -444,9 +444,12 @@ public class SlidingWindowedKStreamIntegrationTest {
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
- consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, windowSize);
- if (keyDeserializer instanceof TimeWindowedDeserializer || keyDeserializer instanceof SessionWindowedDeserializer) {
- consumerProperties.setProperty(StreamsConfig.WINDOWED_INNER_CLASS_SERDE,
+ consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, windowSize);
+ if (keyDeserializer instanceof TimeWindowedDeserializer) {
+ consumerProperties.setProperty(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
+ Serdes.serdeFrom(innerClass).getClass().getName());
+ } else if (keyDeserializer instanceof SessionWindowedDeserializer) {
+ consumerProperties.setProperty(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
Serdes.serdeFrom(innerClass).getClass().getName());
}
return IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TimeWindowedKStreamIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TimeWindowedKStreamIntegrationTest.java
index 14a07e569ee..53746cd7fd6 100644
--- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TimeWindowedKStreamIntegrationTest.java
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TimeWindowedKStreamIntegrationTest.java
@@ -478,9 +478,12 @@ public class TimeWindowedKStreamIntegrationTest {
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
- consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, windowSize);
- if (keyDeserializer instanceof TimeWindowedDeserializer || keyDeserializer instanceof SessionWindowedDeserializer) {
- consumerProperties.setProperty(StreamsConfig.WINDOWED_INNER_CLASS_SERDE,
+ consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, windowSize);
+ if (keyDeserializer instanceof TimeWindowedDeserializer) {
+ consumerProperties.setProperty(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
+ Serdes.serdeFrom(innerClass).getClass().getName());
+ } else if (keyDeserializer instanceof SessionWindowedDeserializer) {
+ consumerProperties.setProperty(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
Serdes.serdeFrom(innerClass).getClass().getName());
}
return IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index 6aedae59576..b98806590a1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -45,6 +45,10 @@ import org.apache.kafka.streams.errors.ProductionExceptionHandler;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.internals.StreamsConfigUtils;
import org.apache.kafka.streams.internals.UpgradeFromValues;
+import org.apache.kafka.streams.kstream.SessionWindowedDeserializer;
+import org.apache.kafka.streams.kstream.SessionWindowedSerializer;
+import org.apache.kafka.streams.kstream.TimeWindowedDeserializer;
+import org.apache.kafka.streams.kstream.TimeWindowedSerializer;
import org.apache.kafka.streams.processor.FailOnInvalidTimestamp;
import org.apache.kafka.streams.processor.TimestampExtractor;
import org.apache.kafka.streams.processor.assignment.TaskAssignor;
@@ -829,13 +833,28 @@ public class StreamsConfig extends AbstractConfig {
+ CONFIG_ERROR_MSG
+ "\"NO_OPTIMIZATION\" by default.";
- /** {@code windowed.inner.class.serde} */
+ /**
+ * {@code windowed.inner.class.serde}
+ *
+ * @deprecated since 4.1.0.
+ * Use {@link TimeWindowedSerializer#WINDOWED_INNER_SERIALIZER_CLASS} for {@link TimeWindowedSerializer}.
+ * Use {@link TimeWindowedDeserializer#WINDOWED_INNER_DESERIALIZER_CLASS} for {@link TimeWindowedDeserializer}.
+ * Use {@link SessionWindowedSerializer#WINDOWED_INNER_SERIALIZER_CLASS} for {@link SessionWindowedSerializer}.
+ * Use {@link SessionWindowedDeserializer#WINDOWED_INNER_DESERIALIZER_CLASS} for {@link SessionWindowedDeserializer}.
+ */
+ @Deprecated
public static final String WINDOWED_INNER_CLASS_SERDE = "windowed.inner.class.serde";
private static final String WINDOWED_INNER_CLASS_SERDE_DOC = " Default serializer / deserializer for the inner class of a windowed record. Must implement the " +
"<code>org.apache.kafka.common.serialization.Serde</code> interface. Note that setting this config in KafkaStreams application would result " +
"in an error as it is meant to be used only from Plain consumer client.";
- /** {@code window.size.ms} */
+ /**
+ * {@code window.size.ms}
+ *
+ * @deprecated since 4.1.0.
+ * Use {@link TimeWindowedDeserializer#WINDOW_SIZE_MS_CONFIG} for {@link TimeWindowedDeserializer}.
+ */
+ @Deprecated
public static final String WINDOW_SIZE_MS_CONFIG = "window.size.ms";
private static final String WINDOW_SIZE_MS_DOC = "Sets window size for the deserializer in order to calculate window end times.";
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedDeserializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedDeserializer.java
index e77efe799cd..11795459c9c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedDeserializer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedDeserializer.java
@@ -23,10 +23,20 @@ import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.state.internals.SessionKeySchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.Map;
public class SessionWindowedDeserializer<T> implements Deserializer<Windowed<T>> {
+ /**
+ * Default deserializer for the inner deserializer class of a windowed record. Must implement the {@link Serde} interface.
+ */
+ public static final String WINDOWED_INNER_DESERIALIZER_CLASS = "windowed.inner.deserializer.class";
+
+ private final Logger log = LoggerFactory.getLogger(SessionWindowedDeserializer.class);
+
private Deserializer<T> inner;
// Default constructor needed by Kafka
@@ -36,34 +46,43 @@ public class SessionWindowedDeserializer<T> implements Deserializer<Windowed<T>>
this.inner = inner;
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"deprecation", "unchecked"})
@Override
public void configure(final Map<String, ?> configs, final boolean isKey) {
- final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
-
- Serde<T> windowInnerClassSerde = null;
+ String deserializerConfigKey = WINDOWED_INNER_DESERIALIZER_CLASS;
+ String deserializerConfigValue = (String) configs.get(WINDOWED_INNER_DESERIALIZER_CLASS);
+ if (deserializerConfigValue == null) {
+ final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
+ if (windowedInnerClassSerdeConfig != null) {
+ deserializerConfigKey = StreamsConfig.WINDOWED_INNER_CLASS_SERDE;
+ deserializerConfigValue = windowedInnerClassSerdeConfig;
+ log.warn("Config {} is deprecated. Please use {} instead.",
+ StreamsConfig.WINDOWED_INNER_CLASS_SERDE, WINDOWED_INNER_DESERIALIZER_CLASS);
+ }
+ }
- if (windowedInnerClassSerdeConfig != null) {
+ Serde<T> windowedInnerDeserializerClass = null;
+ if (deserializerConfigValue != null) {
try {
- windowInnerClassSerde = Utils.newInstance(windowedInnerClassSerdeConfig, Serde.class);
+ windowedInnerDeserializerClass = Utils.newInstance(deserializerConfigValue, Serde.class);
} catch (final ClassNotFoundException e) {
- throw new ConfigException(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, windowedInnerClassSerdeConfig,
- "Serde class " + windowedInnerClassSerdeConfig + " could not be found.");
+ throw new ConfigException(deserializerConfigKey, deserializerConfigValue,
+ "Serde class " + deserializerConfigValue + " could not be found.");
}
}
- if (inner != null && windowedInnerClassSerdeConfig != null) {
- if (!inner.getClass().getName().equals(windowInnerClassSerde.deserializer().getClass().getName())) {
+ if (inner != null && deserializerConfigValue != null) {
+ if (!inner.getClass().getName().equals(windowedInnerDeserializerClass.deserializer().getClass().getName())) {
throw new IllegalArgumentException("Inner class deserializer set using constructor "
+ "(" + inner.getClass().getName() + ")" +
- " is different from the one set in windowed.inner.class.serde config " +
- "(" + windowInnerClassSerde.deserializer().getClass().getName() + ").");
+ " is different from the one set in " + deserializerConfigKey + " config " +
+ "(" + windowedInnerDeserializerClass.deserializer().getClass().getName() + ").");
}
- } else if (inner == null && windowedInnerClassSerdeConfig == null) {
+ } else if (inner == null && deserializerConfigValue == null) {
throw new IllegalArgumentException("Inner class deserializer should be set either via constructor " +
- "or via the windowed.inner.class.serde config");
+ "or via the " + WINDOWED_INNER_DESERIALIZER_CLASS + " config");
} else if (inner == null)
- inner = windowInnerClassSerde.deserializer();
+ inner = windowedInnerDeserializerClass.deserializer();
}
@Override
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedSerializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedSerializer.java
index 6ec10bf8668..60e2b81d497 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedSerializer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedSerializer.java
@@ -24,10 +24,20 @@ import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.internals.WindowedSerializer;
import org.apache.kafka.streams.state.internals.SessionKeySchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.Map;
public class SessionWindowedSerializer<T> implements WindowedSerializer<T> {
+ /**
+ * Default serializer for the inner serializer class of a windowed record. Must implement the {@link Serde} interface.
+ */
+ public static final String WINDOWED_INNER_SERIALIZER_CLASS = "windowed.inner.serializer.class";
+
+ private final Logger log = LoggerFactory.getLogger(SessionWindowedSerializer.class);
+
private Serializer<T> inner;
// Default constructor needed by Kafka
@@ -37,32 +47,42 @@ public class SessionWindowedSerializer<T> implements WindowedSerializer<T> {
this.inner = inner;
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"deprecation", "unchecked"})
@Override
public void configure(final Map<String, ?> configs, final boolean isKey) {
- final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
- Serde<T> windowInnerClassSerde = null;
- if (windowedInnerClassSerdeConfig != null) {
+ String serializerConfigKey = WINDOWED_INNER_SERIALIZER_CLASS;
+ String serializerConfigValue = (String) configs.get(WINDOWED_INNER_SERIALIZER_CLASS);
+ if (serializerConfigValue == null) {
+ final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
+ if (windowedInnerClassSerdeConfig != null) {
+ serializerConfigKey = StreamsConfig.WINDOWED_INNER_CLASS_SERDE;
+ serializerConfigValue = windowedInnerClassSerdeConfig;
+ log.warn("Config {} is deprecated. Please use {} instead.",
+ StreamsConfig.WINDOWED_INNER_CLASS_SERDE, WINDOWED_INNER_SERIALIZER_CLASS);
+ }
+ }
+ Serde<T> windowedInnerSerializerClass = null;
+ if (serializerConfigValue != null) {
try {
- windowInnerClassSerde = Utils.newInstance(windowedInnerClassSerdeConfig, Serde.class);
+ windowedInnerSerializerClass = Utils.newInstance(serializerConfigValue, Serde.class);
} catch (final ClassNotFoundException e) {
- throw new ConfigException(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, windowedInnerClassSerdeConfig,
- "Serde class " + windowedInnerClassSerdeConfig + " could not be found.");
+ throw new ConfigException(serializerConfigKey, serializerConfigValue,
+ "Serde class " + serializerConfigValue + " could not be found.");
}
}
- if (inner != null && windowedInnerClassSerdeConfig != null) {
- if (!inner.getClass().getName().equals(windowInnerClassSerde.serializer().getClass().getName())) {
+ if (inner != null && serializerConfigValue != null) {
+ if (!inner.getClass().getName().equals(windowedInnerSerializerClass.serializer().getClass().getName())) {
throw new IllegalArgumentException("Inner class serializer set using constructor "
+ "(" + inner.getClass().getName() + ")" +
- " is different from the one set in windowed.inner.class.serde config " +
- "(" + windowInnerClassSerde.serializer().getClass().getName() + ").");
+ " is different from the one set in " + serializerConfigKey + " config " +
+ "(" + windowedInnerSerializerClass.serializer().getClass().getName() + ").");
}
- } else if (inner == null && windowedInnerClassSerdeConfig == null) {
+ } else if (inner == null && serializerConfigValue == null) {
throw new IllegalArgumentException("Inner class serializer should be set either via constructor " +
- "or via the windowed.inner.class.serde config");
+ "or via the " + WINDOWED_INNER_SERIALIZER_CLASS + " config");
} else if (inner == null)
- inner = windowInnerClassSerde.serializer();
+ inner = windowedInnerSerializerClass.serializer();
}
@Override
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializer.java
index 77825f2e1fa..26fcbac785f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializer.java
@@ -23,10 +23,25 @@ import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.state.internals.WindowKeySchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.Map;
public class TimeWindowedDeserializer<T> implements Deserializer<Windowed<T>> {
+ /**
+ * Sets window size for the deserializer in order to calculate window end times.
+ */
+ public static final String WINDOW_SIZE_MS_CONFIG = "window.size.ms";
+
+ /**
+ * Default deserializer for the inner deserializer class of a windowed record. Must implement the {@link Serde} interface.
+ */
+ public static final String WINDOWED_INNER_DESERIALIZER_CLASS = "windowed.inner.deserializer.class";
+
+ private final Logger log = LoggerFactory.getLogger(TimeWindowedDeserializer.class);
+
private Long windowSize;
private boolean isChangelogTopic;
@@ -47,50 +62,10 @@ public class TimeWindowedDeserializer<T> implements Deserializer<Windowed<T>> {
return this.windowSize;
}
- @SuppressWarnings("unchecked")
@Override
public void configure(final Map<String, ?> configs, final boolean isKey) {
- //check to see if the window size config is set and the window size is already set from the constructor
- final Long configWindowSize;
- if (configs.get(StreamsConfig.WINDOW_SIZE_MS_CONFIG) instanceof String) {
- configWindowSize = Long.parseLong((String) configs.get(StreamsConfig.WINDOW_SIZE_MS_CONFIG));
- } else {
- configWindowSize = (Long) configs.get(StreamsConfig.WINDOW_SIZE_MS_CONFIG);
- }
- if (windowSize != null && configWindowSize != null) {
- throw new IllegalArgumentException("Window size should not be set in both the time windowed deserializer constructor and the window.size.ms config");
- } else if (windowSize == null && configWindowSize == null) {
- throw new IllegalArgumentException("Window size needs to be set either through the time windowed deserializer " +
- "constructor or the window.size.ms config but not both");
- } else {
- windowSize = windowSize == null ? configWindowSize : windowSize;
- }
-
- final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
-
- Serde<T> windowInnerClassSerde = null;
-
- if (windowedInnerClassSerdeConfig != null) {
- try {
- windowInnerClassSerde = Utils.newInstance(windowedInnerClassSerdeConfig, Serde.class);
- } catch (final ClassNotFoundException e) {
- throw new ConfigException(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, windowedInnerClassSerdeConfig,
- "Serde class " + windowedInnerClassSerdeConfig + " could not be found.");
- }
- }
-
- if (inner != null && windowedInnerClassSerdeConfig != null) {
- if (!inner.getClass().getName().equals(windowInnerClassSerde.deserializer().getClass().getName())) {
- throw new IllegalArgumentException("Inner class deserializer set using constructor "
- + "(" + inner.getClass().getName() + ")" +
- " is different from the one set in windowed.inner.class.serde config " +
- "(" + windowInnerClassSerde.deserializer().getClass().getName() + ").");
- }
- } else if (inner == null && windowedInnerClassSerdeConfig == null) {
- throw new IllegalArgumentException("Inner class deserializer should be set either via constructor " +
- "or via the windowed.inner.class.serde config");
- } else if (inner == null)
- inner = windowInnerClassSerde.deserializer();
+ configureWindowSizeMs(configs);
+ configureWindowInnerDeserializerClass(configs);
}
@Override
@@ -125,4 +100,60 @@ public class TimeWindowedDeserializer<T> implements Deserializer<Windowed<T>> {
Deserializer<T> innerDeserializer() {
return inner;
}
+
+ private void configureWindowSizeMs(final Map<String, ?> configs) {
+ //check to see if the window size config is set and the window size is already set from the constructor
+ final Long configWindowSize;
+ if (configs.get(WINDOW_SIZE_MS_CONFIG) instanceof String) {
+ configWindowSize = Long.parseLong((String) configs.get(WINDOW_SIZE_MS_CONFIG));
+ } else {
+ configWindowSize = (Long) configs.get(WINDOW_SIZE_MS_CONFIG);
+ }
+ if (windowSize != null && configWindowSize != null) {
+ throw new IllegalArgumentException("Window size should not be set in both the time windowed deserializer constructor and the window.size.ms config");
+ } else if (windowSize == null && configWindowSize == null) {
+ throw new IllegalArgumentException("Window size needs to be set either through the time windowed deserializer " +
+ "constructor or the window.size.ms config but not both");
+ } else {
+ windowSize = windowSize == null ? configWindowSize : windowSize;
+ }
+ }
+
+ @SuppressWarnings({"deprecation", "unchecked"})
+ private void configureWindowInnerDeserializerClass(final Map<String, ?> configs) {
+ String deserializerConfigKey = WINDOWED_INNER_DESERIALIZER_CLASS;
+ String deserializerConfigValue = (String) configs.get(WINDOWED_INNER_DESERIALIZER_CLASS);
+ if (deserializerConfigValue == null) {
+ final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
+ if (windowedInnerClassSerdeConfig != null) {
+ deserializerConfigKey = StreamsConfig.WINDOWED_INNER_CLASS_SERDE;
+ deserializerConfigValue = windowedInnerClassSerdeConfig;
+ log.warn("Config {} is deprecated. Please use {} instead.",
+ StreamsConfig.WINDOWED_INNER_CLASS_SERDE, WINDOWED_INNER_DESERIALIZER_CLASS);
+ }
+ }
+
+ Serde<T> windowedInnerDeserializerClass = null;
+ if (deserializerConfigValue != null) {
+ try {
+ windowedInnerDeserializerClass = Utils.newInstance(deserializerConfigValue, Serde.class);
+ } catch (final ClassNotFoundException e) {
+ throw new ConfigException(deserializerConfigKey, deserializerConfigValue,
+ "Serde class " + deserializerConfigValue + " could not be found.");
+ }
+ }
+
+ if (inner != null && deserializerConfigValue != null) {
+ if (!inner.getClass().getName().equals(windowedInnerDeserializerClass.deserializer().getClass().getName())) {
+ throw new IllegalArgumentException("Inner class deserializer set using constructor "
+ + "(" + inner.getClass().getName() + ")" +
+ " is different from the one set in " + deserializerConfigKey + " config " +
+ "(" + windowedInnerDeserializerClass.deserializer().getClass().getName() + ").");
+ }
+ } else if (inner == null && deserializerConfigValue == null) {
+ throw new IllegalArgumentException("Inner class deserializer should be set either via constructor " +
+ "or via the " + WINDOWED_INNER_DESERIALIZER_CLASS + " config");
+ } else if (inner == null)
+ inner = windowedInnerDeserializerClass.deserializer();
+ }
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindowedSerializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindowedSerializer.java
index 54bdd6a3b71..7cd13afc1d3 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindowedSerializer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindowedSerializer.java
@@ -24,10 +24,20 @@ import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.internals.WindowedSerializer;
import org.apache.kafka.streams.state.internals.WindowKeySchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.Map;
public class TimeWindowedSerializer<T> implements WindowedSerializer<T> {
+ /**
+ * Default serializer for the inner serializer class of a windowed record. Must implement the {@link Serde} interface.
+ */
+ public static final String WINDOWED_INNER_SERIALIZER_CLASS = "windowed.inner.serializer.class";
+
+ private final Logger log = LoggerFactory.getLogger(TimeWindowedSerializer.class);
+
private Serializer<T> inner;
// Default constructor needed by Kafka
@@ -38,32 +48,42 @@ public class TimeWindowedSerializer<T> implements WindowedSerializer<T> {
this.inner = inner;
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"deprecation", "unchecked"})
@Override
public void configure(final Map<String, ?> configs, final boolean isKey) {
- final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
- Serde<T> windowInnerClassSerde = null;
- if (windowedInnerClassSerdeConfig != null) {
+ String serializerConfigKey = WINDOWED_INNER_SERIALIZER_CLASS;
+ String serializerConfigValue = (String) configs.get(WINDOWED_INNER_SERIALIZER_CLASS);
+ if (serializerConfigValue == null) {
+ final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
+ if (windowedInnerClassSerdeConfig != null) {
+ serializerConfigKey = StreamsConfig.WINDOWED_INNER_CLASS_SERDE;
+ serializerConfigValue = windowedInnerClassSerdeConfig;
+ log.warn("Config {} is deprecated. Please use {} instead.",
+ StreamsConfig.WINDOWED_INNER_CLASS_SERDE, WINDOWED_INNER_SERIALIZER_CLASS);
+ }
+ }
+ Serde<T> windowedInnerSerializerClass = null;
+ if (serializerConfigValue != null) {
try {
- windowInnerClassSerde = Utils.newInstance(windowedInnerClassSerdeConfig, Serde.class);
+ windowedInnerSerializerClass = Utils.newInstance(serializerConfigValue, Serde.class);
} catch (final ClassNotFoundException e) {
- throw new ConfigException(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, windowedInnerClassSerdeConfig,
- "Serde class " + windowedInnerClassSerdeConfig + " could not be found.");
+ throw new ConfigException(serializerConfigKey, serializerConfigValue,
+ "Serde class " + serializerConfigValue + " could not be found.");
}
}
- if (inner != null && windowedInnerClassSerdeConfig != null) {
- if (!inner.getClass().getName().equals(windowInnerClassSerde.serializer().getClass().getName())) {
+ if (inner != null && serializerConfigValue != null) {
+ if (!inner.getClass().getName().equals(windowedInnerSerializerClass.serializer().getClass().getName())) {
throw new IllegalArgumentException("Inner class serializer set using constructor "
+ "(" + inner.getClass().getName() + ")" +
- " is different from the one set in windowed.inner.class.serde config " +
- "(" + windowInnerClassSerde.serializer().getClass().getName() + ").");
+ " is different from the one set in " + serializerConfigKey + " config " +
+ "(" + windowedInnerSerializerClass.serializer().getClass().getName() + ").");
}
- } else if (inner == null && windowedInnerClassSerdeConfig == null) {
+ } else if (inner == null && serializerConfigValue == null) {
throw new IllegalArgumentException("Inner class serializer should be set either via constructor " +
- "or via the windowed.inner.class.serde config");
+ "or via the " + WINDOWED_INNER_SERIALIZER_CLASS + " config");
} else if (inner == null)
- inner = windowInnerClassSerde.serializer();
+ inner = windowedInnerSerializerClass.serializer();
}
@Override
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowedDeserializerTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowedDeserializerTest.java
index 82f0e355307..2bace901779 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowedDeserializerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowedDeserializerTest.java
@@ -45,7 +45,7 @@ public class SessionWindowedDeserializerTest {
}
@Test
- public void shouldSetWindowedInnerClassDeserialiserThroughConfig() {
+ public void shouldSetSerializerThroughWindowedInnerClassSerdeConfig() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
final SessionWindowedDeserializer<?> deserializer = new SessionWindowedDeserializer<>();
deserializer.configure(props, false);
@@ -53,20 +53,49 @@ public class SessionWindowedDeserializerTest {
}
@Test
- public void shouldThrowErrorIfWindowInnerClassDeserialiserIsNotSet() {
+ public void shouldSetSerializerThroughWindowedInnerDeserializerClassConfig() {
+ props.put(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ final SessionWindowedDeserializer<?> deserializer = new SessionWindowedDeserializer<>();
+ deserializer.configure(props, false);
+ assertInstanceOf(ByteArrayDeserializer.class, deserializer.innerDeserializer());
+ }
+
+ @Test
+ public void shouldIgnoreWindowedInnerClassSerdeConfigIfWindowedInnerDeserializerClassConfigIsSet() {
+ props.put(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, "some.non.existent.class");
+ final SessionWindowedDeserializer<?> deserializer = new SessionWindowedDeserializer<>();
+ deserializer.configure(props, false);
+ assertInstanceOf(ByteArrayDeserializer.class, deserializer.innerDeserializer());
+ }
+
+ @Test
+ public void shouldThrowErrorIfWindowedInnerClassSerdeAndSessionWindowedDeserializerClassAreNotSet() {
final SessionWindowedDeserializer<?> deserializer = new SessionWindowedDeserializer<>();
assertThrows(IllegalArgumentException.class, () -> deserializer.configure(props, false));
}
@Test
- public void shouldThrowErrorIfDeserialisersConflictInConstructorAndConfig() {
+ public void shouldThrowErrorIfDeserializersConflictInConstructorAndWindowedInnerClassSerdeConfig() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
assertThrows(IllegalArgumentException.class, () -> sessionWindowedDeserializer.configure(props, false));
}
@Test
- public void shouldThrowConfigExceptionWhenInvalidWindowInnerClassDeserialiserSupplied() {
+ public void shouldThrowErrorIfDeserializersConflictInConstructorAndWindowedInnerDeserializerClassConfig() {
+ props.put(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ assertThrows(IllegalArgumentException.class, () -> sessionWindowedDeserializer.configure(props, false));
+ }
+
+ @Test
+ public void shouldThrowConfigExceptionWhenInvalidWindowedInnerClassSerdeSupplied() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, "some.non.existent.class");
assertThrows(ConfigException.class, () -> sessionWindowedDeserializer.configure(props, false));
}
+
+ @Test
+ public void shouldThrowConfigExceptionWhenInvalidWindowedInnerDeserializerClassSupplied() {
+ props.put(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS, "some.non.existent.class");
+ assertThrows(ConfigException.class, () -> sessionWindowedDeserializer.configure(props, false));
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowedSerializerTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowedSerializerTest.java
index 729c8408dfe..d7e30bc3fe4 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowedSerializerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowedSerializerTest.java
@@ -45,7 +45,7 @@ public class SessionWindowedSerializerTest {
}
@Test
- public void shouldSetWindowedInnerClassSerialiserThroughConfig() {
+ public void shouldSetSerializerThroughWindowedInnerClassSerdeConfig() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
final SessionWindowedSerializer<?> serializer = new SessionWindowedSerializer<>();
serializer.configure(props, false);
@@ -53,20 +53,49 @@ public class SessionWindowedSerializerTest {
}
@Test
- public void shouldThrowErrorIfWindowInnerClassSerialiserIsNotSet() {
+ public void shouldSetSerializerThroughWindowedInnerSerializerClassConfig() {
+ props.put(SessionWindowedSerializer.WINDOWED_INNER_SERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ final SessionWindowedSerializer<?> serializer = new SessionWindowedSerializer<>();
+ serializer.configure(props, false);
+ assertInstanceOf(ByteArraySerializer.class, serializer.innerSerializer());
+ }
+
+ @Test
+ public void shouldIgnoreWindowedInnerClassSerdeConfigIfWindowedInnerSerializerClassConfigIsSet() {
+ props.put(SessionWindowedSerializer.WINDOWED_INNER_SERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, "some.non.existent.class");
+ final SessionWindowedSerializer<?> serializer = new SessionWindowedSerializer<>();
+ serializer.configure(props, false);
+ assertInstanceOf(ByteArraySerializer.class, serializer.innerSerializer());
+ }
+
+ @Test
+ public void shouldThrowErrorIfWindowedInnerClassSerdeAndWindowedInnerSerializerClassAreNotSet() {
final SessionWindowedSerializer<?> serializer = new SessionWindowedSerializer<>();
assertThrows(IllegalArgumentException.class, () -> serializer.configure(props, false));
}
@Test
- public void shouldThrowErrorIfSerialisersConflictInConstructorAndConfig() {
+ public void shouldThrowErrorIfSerializersConflictInConstructorAndWindowedInnerClassSerdeConfig() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
assertThrows(IllegalArgumentException.class, () -> sessionWindowedSerializer.configure(props, false));
}
@Test
- public void shouldThrowConfigExceptionWhenInvalidWindowInnerClassSerialiserSupplied() {
+ public void shouldThrowErrorIfSerializersConflictInConstructorAndWindowedInnerSerializerClassConfig() {
+ props.put(SessionWindowedSerializer.WINDOWED_INNER_SERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ assertThrows(IllegalArgumentException.class, () -> sessionWindowedSerializer.configure(props, false));
+ }
+
+ @Test
+ public void shouldThrowConfigExceptionWhenInvalidWindowedInnerClassSerdeSupplied() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, "some.non.existent.class");
assertThrows(ConfigException.class, () -> sessionWindowedSerializer.configure(props, false));
}
+
+ @Test
+ public void shouldThrowConfigExceptionWhenInvalidWindowedInnerSerializerClassSupplied() {
+ props.put(SessionWindowedSerializer.WINDOWED_INNER_SERIALIZER_CLASS, "some.non.existent.class");
+ assertThrows(ConfigException.class, () -> sessionWindowedSerializer.configure(props, false));
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializerTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializerTest.java
index b82b48d7730..bfb8c80cf09 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowedDeserializerTest.java
@@ -30,6 +30,7 @@ import java.util.Map;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.Is.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -49,7 +50,7 @@ public class TimeWindowedDeserializerTest {
}
@Test
- public void shouldSetWindowSizeAndWindowedInnerDeserialiserThroughConfigs() {
+ public void shouldSetWindowSizeAndDeserializerThroughWindowSizeMsAndWindowedInnerClassSerdeConfigs() {
props.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, "500");
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
final TimeWindowedDeserializer<?> deserializer = new TimeWindowedDeserializer<>();
@@ -59,34 +60,92 @@ public class TimeWindowedDeserializerTest {
}
@Test
- public void shouldThrowErrorIfWindowSizeSetInConfigsAndConstructor() {
+ public void shouldSetWindowSizeAndDeserializerThroughWindowSizeMsAndWindowedInnerDeserializerClassConfigs() {
+ props.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, "500");
+ props.put(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ final TimeWindowedDeserializer<?> deserializer = new TimeWindowedDeserializer<>();
+ deserializer.configure(props, false);
+ assertThat(deserializer.getWindowSize(), is(500L));
+ assertInstanceOf(ByteArrayDeserializer.class, deserializer.innerDeserializer());
+ }
+
+ @Test
+ public void shouldHaveSameConfigNameForWindowSizeMs() {
+ assertEquals(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, StreamsConfig.WINDOW_SIZE_MS_CONFIG);
+ }
+
+ @Test
+ public void shouldIgnoreWindowedInnerClassSerdeConfigIfWindowedInnerDeserializerClassConfigIsSet() {
+ props.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, "500");
+ props.put(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, "some.non.existent.class");
+ final TimeWindowedDeserializer<?> deserializer = new TimeWindowedDeserializer<>();
+ deserializer.configure(props, false);
+ assertThat(deserializer.getWindowSize(), is(500L));
+ assertInstanceOf(ByteArrayDeserializer.class, deserializer.innerDeserializer());
+ }
+
+ @Test
+ public void shouldThrowErrorIfWindowSizeSetInStreamsConfigAndConstructor() {
props.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, "500");
assertThrows(IllegalArgumentException.class, () -> timeWindowedDeserializer.configure(props, false));
}
@Test
- public void shouldThrowErrorIfWindowSizeIsNotSet() {
+ public void shouldThrowErrorIfWindowSizeSetInConstructorConfigAndConstructor() {
+ props.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, "500");
+ assertThrows(IllegalArgumentException.class, () -> timeWindowedDeserializer.configure(props, false));
+ }
+
+ @Test
+ public void shouldThrowErrorIfWindowSizeIsNotSetAndWindowedInnerClassSerdeIsSet() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
final TimeWindowedDeserializer<?> deserializer = new TimeWindowedDeserializer<>();
assertThrows(IllegalArgumentException.class, () -> deserializer.configure(props, false));
}
@Test
- public void shouldThrowErrorIfWindowedInnerClassDeserialiserIsNotSet() {
+ public void shouldThrowErrorIfWindowSizeIsNotSetAndWindowedInnerDeserializerClassIsSet() {
+ props.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, Serdes.ByteArraySerde.class.getName());
+ final TimeWindowedDeserializer<?> deserializer = new TimeWindowedDeserializer<>();
+ assertThrows(IllegalArgumentException.class, () -> deserializer.configure(props, false));
+ }
+
+ @Test
+ public void shouldThrowErrorIfWindowedInnerClassSerdeIsNotSetAndWindowSizeMsInStreamsConfigIsSet() {
props.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, "500");
final TimeWindowedDeserializer<?> deserializer = new TimeWindowedDeserializer<>();
assertThrows(IllegalArgumentException.class, () -> deserializer.configure(props, false));
}
@Test
- public void shouldThrowErrorIfWindowedInnerClassDeserialisersConflictInConstructorAndConfig() {
+ public void shouldThrowErrorIfWindowedInnerClassSerdeIsNotSetAndWindowSizeMsInConstructorConfigIsSet() {
+ props.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, "500");
+ final TimeWindowedDeserializer<?> deserializer = new TimeWindowedDeserializer<>();
+ assertThrows(IllegalArgumentException.class, () -> deserializer.configure(props, false));
+ }
+
+ @Test
+ public void shouldThrowErrorIfDeserializerConflictInConstructorAndWindowedInnerClassSerdeConfig() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
assertThrows(IllegalArgumentException.class, () -> timeWindowedDeserializer.configure(props, false));
}
@Test
- public void shouldThrowConfigExceptionWhenInvalidWindowedInnerClassDeserialiserSupplied() {
+ public void shouldThrowErrorIfDeserializerConflictInConstructorAndWindowedInnerDeserializerClassConfig() {
+ props.put(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ assertThrows(IllegalArgumentException.class, () -> timeWindowedDeserializer.configure(props, false));
+ }
+
+ @Test
+ public void shouldThrowConfigExceptionWhenInvalidWindowedInnerClassSerdeSupplied() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, "some.non.existent.class");
assertThrows(ConfigException.class, () -> timeWindowedDeserializer.configure(props, false));
}
+
+ @Test
+ public void shouldThrowConfigExceptionWhenInvalidWindowedInnerDeserializerClassSupplied() {
+ props.put(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS, "some.non.existent.class");
+ assertThrows(ConfigException.class, () -> timeWindowedDeserializer.configure(props, false));
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowedSerializerTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowedSerializerTest.java
index ba5b9c339a4..7a13117db4a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowedSerializerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowedSerializerTest.java
@@ -45,7 +45,7 @@ public class TimeWindowedSerializerTest {
}
@Test
- public void shouldSetWindowedInnerClassSerialiserThroughConfig() {
+ public void shouldSetSerializerThroughWindowedInnerClassSerdeConfig() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
final TimeWindowedSerializer<?> serializer = new TimeWindowedSerializer<>();
serializer.configure(props, false);
@@ -53,21 +53,49 @@ public class TimeWindowedSerializerTest {
}
@Test
- public void shouldThrowErrorIfWindowedInnerClassSerialiserIsNotSet() {
+ public void shouldSetSerializerThroughWindowedInnerSerializerClassConfig() {
+ props.put(TimeWindowedSerializer.WINDOWED_INNER_SERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ final TimeWindowedSerializer<?> serializer = new TimeWindowedSerializer<>();
+ serializer.configure(props, false);
+ assertInstanceOf(ByteArraySerializer.class, serializer.innerSerializer());
+ }
+
+ @Test
+ public void shouldIgnoreWindowedInnerClassSerdeConfigIfWindowedInnerSerializerClassConfigIsSet() {
+ props.put(TimeWindowedSerializer.WINDOWED_INNER_SERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, "some.non.existent.class");
+ final TimeWindowedSerializer<?> serializer = new TimeWindowedSerializer<>();
+ serializer.configure(props, false);
+ assertInstanceOf(ByteArraySerializer.class, serializer.innerSerializer());
+ }
+
+ @Test
+ public void shouldThrowErrorIfWindowedInnerClassSerdeAndWindowedInnerSerializerClassAreNotSet() {
final TimeWindowedSerializer<?> serializer = new TimeWindowedSerializer<>();
assertThrows(IllegalArgumentException.class, () -> serializer.configure(props, false));
}
@Test
- public void shouldThrowErrorIfWindowedInnerClassSerialisersConflictInConstructorAndConfig() {
+ public void shouldThrowErrorIfSerializerConflictInConstructorAndWindowedInnerClassSerdeConfig() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, Serdes.ByteArraySerde.class.getName());
assertThrows(IllegalArgumentException.class, () -> timeWindowedSerializer.configure(props, false));
}
@Test
- public void shouldThrowConfigExceptionWhenInvalidWindowedInnerClassSerialiserSupplied() {
+ public void shouldThrowErrorIfSerializerConflictInConstructorAndWindowedInnerSerializerClassConfig() {
+ props.put(TimeWindowedSerializer.WINDOWED_INNER_SERIALIZER_CLASS, Serdes.ByteArraySerde.class.getName());
+ assertThrows(IllegalArgumentException.class, () -> timeWindowedSerializer.configure(props, false));
+ }
+
+ @Test
+ public void shouldThrowConfigExceptionWhenInvalidWindowedInnerClassSerdeSupplied() {
props.put(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, "some.non.existent.class");
assertThrows(ConfigException.class, () -> timeWindowedSerializer.configure(props, false));
}
+ @Test
+ public void shouldThrowConfigExceptionWhenInvalidWindowedInnerSerializerClassSupplied() {
+ props.put(TimeWindowedSerializer.WINDOWED_INNER_SERIALIZER_CLASS, "some.non.existent.class");
+ assertThrows(ConfigException.class, () -> timeWindowedSerializer.configure(props, false));
+ }
}