KAFKA-16260: Deprecate window.size.ms and window.inner.class.serde in StreamsConfig #18297
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -1065,9 +1065,12 @@ private <K, V> List<KeyValueTimestamp<K, V>> receiveMessages(final Deserializer< | |
| 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(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 @@ private <K, V> List<KeyValueTimestamp<K, V>> receiveMessagesWithTimestamp(final | |
| 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(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 @@ private <K, V> String readWindowedKeyedMessagesViaConsoleConsumer(final Deserial | |
| "--property", "key.deserializer=" + keyDeserializer.getClass().getName(), | ||
| "--property", "value.deserializer=" + valueDeserializer.getClass().getName(), | ||
| "--property", "key.separator=" + keySeparator, | ||
| "--property", "key.deserializer." + TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS + "=" + Serdes.serdeFrom(innerClass).getClass().getName(), | ||
| "--property", "key.deserializer.window.size.ms=500", | ||
| }; | ||
|
|
||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -32,6 +32,7 @@ | |
| 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 @@ private <K, V> boolean processKeyValueAndVerifyCount( | |
| 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(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, 500L); | ||
|
|
||
|
|
||
| final List<KeyValueTimestamp<K, V>> actual = | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -444,9 +444,12 @@ private <K, V> List<KeyValueTimestamp<K, V>> receiveMessagesWithTimestamp(final | |
| 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(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( | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -478,9 +478,12 @@ private <K, V> List<KeyValueTimestamp<K, V>> receiveMessagesWithTimestamp(final | |
| 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(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( | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -45,6 +45,10 @@ | |
| 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; | ||
|
|
@@ -823,13 +827,28 @@ public class StreamsConfig extends AbstractConfig { | |
| + CONFIG_ERROR_MSG | ||
| + "\"NO_OPTIMIZATION\" by default."; | ||
|
|
||
| /** | ||
| * {@code windowed.inner.class.serde} | ||
| * | ||
| * @deprecated since 4.0.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} | ||
| * | ||
| * @deprecated since 4.0.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."; | ||
|
|
||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -23,10 +23,20 @@ | |
| 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>> { | ||
|
|
||
| /** | ||
| * Configuration key for the windowed inner deserializer class. | ||
|
There was a problem hiding this comment. Can we put any more information into this (and the other) java doc comments about the constants? Since this is a public string constant now, the use of the constant should ideally be clear from the javadoc comments. |
||
| */ | ||
| 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 SessionWindowedDeserializer(final Deserializer<T> inner) { | |
| this.inner = inner; | ||
| } | ||
|
|
||
| @SuppressWarnings({"deprecation", "unchecked"}) | ||
| @Override | ||
| public void configure(final Map<String, ?> configs, final boolean isKey) { | ||
| String deserializerConfigFrom = WINDOWED_INNER_DESERIALIZER_CLASS; | ||
| String windowedInnerDeserializerClassConfig = (String) configs.get(WINDOWED_INNER_DESERIALIZER_CLASS); | ||
| if (windowedInnerDeserializerClassConfig == null) { | ||
| final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE); | ||
| if (windowedInnerClassSerdeConfig != null) { | ||
| deserializerConfigFrom = StreamsConfig.WINDOWED_INNER_CLASS_SERDE; | ||
| windowedInnerDeserializerClassConfig = windowedInnerClassSerdeConfig; | ||
| log.warn("Config {} is deprecated. Please use {} instead.", | ||
| StreamsConfig.WINDOWED_INNER_CLASS_SERDE, WINDOWED_INNER_DESERIALIZER_CLASS); | ||
| } | ||
| } | ||
|
|
||
| Serde<T> windowedInnerDeserializerClass = null; | ||
| if (windowedInnerDeserializerClassConfig != null) { | ||
| try { | ||
| windowedInnerDeserializerClass = Utils.newInstance(windowedInnerDeserializerClassConfig, Serde.class); | ||
| } catch (final ClassNotFoundException e) { | ||
| throw new ConfigException(deserializerConfigFrom, windowedInnerDeserializerClassConfig, | ||
| "Serde class " + windowedInnerDeserializerClassConfig + " could not be found."); | ||
| } | ||
| } | ||
|
|
||
| if (inner != null && windowedInnerDeserializerClassConfig != 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 " + deserializerConfigFrom + " config " + | ||
| "(" + windowedInnerDeserializerClass.deserializer().getClass().getName() + ")."); | ||
| } | ||
| } else if (inner == null && windowedInnerDeserializerClassConfig == 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(); | ||
| } | ||
|
|
||
| @Override | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -24,10 +24,20 @@ | |
| 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> { | ||
|
|
||
| /** | ||
| * Configuration key for the windowed inner serializer class. | ||
| */ | ||
| 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 SessionWindowedSerializer(final Serializer<T> inner) { | |
| this.inner = inner; | ||
| } | ||
|
|
||
| @SuppressWarnings({"deprecation", "unchecked"}) | ||
| @Override | ||
| public void configure(final Map<String, ?> configs, final boolean isKey) { | ||
| String serializerConfigFrom = WINDOWED_INNER_SERIALIZER_CLASS; | ||
| String windowedInnerSerializerClassConfig = (String) configs.get(WINDOWED_INNER_SERIALIZER_CLASS); | ||
| if (windowedInnerSerializerClassConfig == null) { | ||
| final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE); | ||
| if (windowedInnerClassSerdeConfig != null) { | ||
| serializerConfigFrom = StreamsConfig.WINDOWED_INNER_CLASS_SERDE; | ||
| windowedInnerSerializerClassConfig = windowedInnerClassSerdeConfig; | ||
| log.warn("Config {} is deprecated. Please use {} instead.", | ||
| StreamsConfig.WINDOWED_INNER_CLASS_SERDE, WINDOWED_INNER_SERIALIZER_CLASS); | ||
| } | ||
| } | ||
| Serde<T> windowedInnerSerializerClass = null; | ||
| if (windowedInnerSerializerClassConfig != null) { | ||
| try { | ||
| windowedInnerSerializerClass = Utils.newInstance(windowedInnerSerializerClassConfig, Serde.class); | ||
| } catch (final ClassNotFoundException e) { | ||
| throw new ConfigException(serializerConfigFrom, windowedInnerSerializerClassConfig, | ||
| "Serde class " + windowedInnerSerializerClassConfig + " could not be found."); | ||
| } | ||
| } | ||
|
|
||
| if (inner != null && windowedInnerSerializerClassConfig != 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 " + serializerConfigFrom + " config " + | ||
| "(" + windowedInnerSerializerClass.serializer().getClass().getName() + ")."); | ||
| } | ||
| } else if (inner == null && windowedInnerSerializerClassConfig == null) { | ||
| throw new IllegalArgumentException("Inner class serializer should be set either via constructor " + | ||
| "or via the " + WINDOWED_INNER_SERIALIZER_CLASS + " config"); | ||
| } else if (inner == null) | ||
| inner = windowedInnerSerializerClass.serializer(); | ||
| } | ||
|
|
||
| @Override | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This should not be in the 4.0 section.