Skip to content

Commit c5f8ae0

Browse files
FrankYang0529FrankYang0529
authored
KAFKA-16260: Deprecate window.size.ms and window.inner.class.serde in StreamsConfig (apache#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 <[email protected]> Signed-off-by: PoAn Yang <[email protected]>
1 parent 6d72677 commit c5f8ae0

14 files changed

+394
-119
lines changed

‎docs/upgrade.html

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,14 @@ <h6><a id="upgrade_410_notable" href="#upgrade_410_notable">Notable changes in 4
3636
</li>
3737
</ul>
3838
</li>
39+
<li><b>Stream</b>
40+
<ul>
41+
<li>
42+
The <code>window.size.ms</code> and <code>window.inner.serde.class</code> in <code>StreamsConfig</code> are deprecated.
43+
Use the corresponding string constants defined in <code>TimeWindowedSerializer</code>, <code>TimeWindowedDeserializer</code>, <code>SessionWindowedSerializer</code> and <code>SessionWindowedDeserializer</code> instead.
44+
</li>
45+
</ul>
46+
</li>
3947
</ul>
4048

4149
<h4><a id="upgrade_4_0_0" href="#upgrade_4_0_0">Upgrading to 4.0.0</a></h4>

‎streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1065,9 +1065,12 @@ private <K, V> List<KeyValueTimestamp<K, V>> receiveMessages(final Deserializer<
10651065
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
10661066
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
10671067
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
1068-
consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, 500L);
1069-
if (keyDeserializer instanceof TimeWindowedDeserializer || keyDeserializer instanceof SessionWindowedDeserializer) {
1070-
consumerProperties.setProperty(StreamsConfig.WINDOWED_INNER_CLASS_SERDE,
1068+
consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, 500L);
1069+
if (keyDeserializer instanceof TimeWindowedDeserializer) {
1070+
consumerProperties.setProperty(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
1071+
Serdes.serdeFrom(innerClass).getClass().getName());
1072+
} else if (keyDeserializer instanceof SessionWindowedDeserializer) {
1073+
consumerProperties.setProperty(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
10711074
Serdes.serdeFrom(innerClass).getClass().getName());
10721075
}
10731076
return IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(
@@ -1089,9 +1092,12 @@ private <K, V> List<KeyValueTimestamp<K, V>> receiveMessagesWithTimestamp(final
10891092
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
10901093
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
10911094
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
1092-
consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, 500L);
1093-
if (keyDeserializer instanceof TimeWindowedDeserializer || keyDeserializer instanceof SessionWindowedDeserializer) {
1094-
consumerProperties.setProperty(StreamsConfig.WINDOWED_INNER_CLASS_SERDE,
1095+
consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, 500L);
1096+
if (keyDeserializer instanceof TimeWindowedDeserializer) {
1097+
consumerProperties.setProperty(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
1098+
Serdes.serdeFrom(innerClass).getClass().getName());
1099+
} else if (keyDeserializer instanceof SessionWindowedDeserializer) {
1100+
consumerProperties.setProperty(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
10951101
Serdes.serdeFrom(innerClass).getClass().getName());
10961102
}
10971103
return IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(
@@ -1123,7 +1129,7 @@ private <K, V> String readWindowedKeyedMessagesViaConsoleConsumer(final Deserial
11231129
"--property", "key.deserializer=" + keyDeserializer.getClass().getName(),
11241130
"--property", "value.deserializer=" + valueDeserializer.getClass().getName(),
11251131
"--property", "key.separator=" + keySeparator,
1126-
"--property", "key.deserializer." + StreamsConfig.WINDOWED_INNER_CLASS_SERDE + "=" + Serdes.serdeFrom(innerClass).getClass().getName(),
1132+
"--property", "key.deserializer." + TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS + "=" + Serdes.serdeFrom(innerClass).getClass().getName(),
11271133
"--property", "key.deserializer.window.size.ms=500",
11281134
};
11291135

‎streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SelfJoinUpgradeIntegrationTest.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import org.apache.kafka.streams.kstream.JoinWindows;
3333
import org.apache.kafka.streams.kstream.KStream;
3434
import org.apache.kafka.streams.kstream.Produced;
35+
import org.apache.kafka.streams.kstream.TimeWindowedDeserializer;
3536
import org.apache.kafka.streams.kstream.ValueJoiner;
3637
import org.apache.kafka.test.TestUtils;
3738

@@ -262,7 +263,7 @@ private <K, V> boolean processKeyValueAndVerifyCount(
262263
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
263264
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
264265
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
265-
consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, 500L);
266+
consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, 500L);
266267

267268

268269
final List<KeyValueTimestamp<K, V>> actual =

‎streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SlidingWindowedKStreamIntegrationTest.java

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -444,9 +444,12 @@ private <K, V> List<KeyValueTimestamp<K, V>> receiveMessagesWithTimestamp(final
444444
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
445445
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
446446
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
447-
consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, windowSize);
448-
if (keyDeserializer instanceof TimeWindowedDeserializer || keyDeserializer instanceof SessionWindowedDeserializer) {
449-
consumerProperties.setProperty(StreamsConfig.WINDOWED_INNER_CLASS_SERDE,
447+
consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, windowSize);
448+
if (keyDeserializer instanceof TimeWindowedDeserializer) {
449+
consumerProperties.setProperty(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
450+
Serdes.serdeFrom(innerClass).getClass().getName());
451+
} else if (keyDeserializer instanceof SessionWindowedDeserializer) {
452+
consumerProperties.setProperty(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
450453
Serdes.serdeFrom(innerClass).getClass().getName());
451454
}
452455
return IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(

‎streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TimeWindowedKStreamIntegrationTest.java

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -478,9 +478,12 @@ private <K, V> List<KeyValueTimestamp<K, V>> receiveMessagesWithTimestamp(final
478478
consumerProperties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
479479
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass().getName());
480480
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer.getClass().getName());
481-
consumerProperties.put(StreamsConfig.WINDOW_SIZE_MS_CONFIG, windowSize);
482-
if (keyDeserializer instanceof TimeWindowedDeserializer || keyDeserializer instanceof SessionWindowedDeserializer) {
483-
consumerProperties.setProperty(StreamsConfig.WINDOWED_INNER_CLASS_SERDE,
481+
consumerProperties.put(TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG, windowSize);
482+
if (keyDeserializer instanceof TimeWindowedDeserializer) {
483+
consumerProperties.setProperty(TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
484+
Serdes.serdeFrom(innerClass).getClass().getName());
485+
} else if (keyDeserializer instanceof SessionWindowedDeserializer) {
486+
consumerProperties.setProperty(SessionWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS,
484487
Serdes.serdeFrom(innerClass).getClass().getName());
485488
}
486489
return IntegrationTestUtils.waitUntilMinKeyValueWithTimestampRecordsReceived(

‎streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,10 @@
4545
import org.apache.kafka.streams.errors.StreamsException;
4646
import org.apache.kafka.streams.internals.StreamsConfigUtils;
4747
import org.apache.kafka.streams.internals.UpgradeFromValues;
48+
import org.apache.kafka.streams.kstream.SessionWindowedDeserializer;
49+
import org.apache.kafka.streams.kstream.SessionWindowedSerializer;
50+
import org.apache.kafka.streams.kstream.TimeWindowedDeserializer;
51+
import org.apache.kafka.streams.kstream.TimeWindowedSerializer;
4852
import org.apache.kafka.streams.processor.FailOnInvalidTimestamp;
4953
import org.apache.kafka.streams.processor.TimestampExtractor;
5054
import org.apache.kafka.streams.processor.assignment.TaskAssignor;
@@ -829,13 +833,28 @@ public class StreamsConfig extends AbstractConfig {
829833
+ CONFIG_ERROR_MSG
830834
+ "\"NO_OPTIMIZATION\" by default.";
831835

832-
/** {@code windowed.inner.class.serde} */
836+
/**
837+
* {@code windowed.inner.class.serde}
838+
*
839+
* @deprecated since 4.1.0.
840+
* Use {@link TimeWindowedSerializer#WINDOWED_INNER_SERIALIZER_CLASS} for {@link TimeWindowedSerializer}.
841+
* Use {@link TimeWindowedDeserializer#WINDOWED_INNER_DESERIALIZER_CLASS} for {@link TimeWindowedDeserializer}.
842+
* Use {@link SessionWindowedSerializer#WINDOWED_INNER_SERIALIZER_CLASS} for {@link SessionWindowedSerializer}.
843+
* Use {@link SessionWindowedDeserializer#WINDOWED_INNER_DESERIALIZER_CLASS} for {@link SessionWindowedDeserializer}.
844+
*/
845+
@Deprecated
833846
public static final String WINDOWED_INNER_CLASS_SERDE = "windowed.inner.class.serde";
834847
private static final String WINDOWED_INNER_CLASS_SERDE_DOC = " Default serializer / deserializer for the inner class of a windowed record. Must implement the " +
835848
"<code>org.apache.kafka.common.serialization.Serde</code> interface. Note that setting this config in KafkaStreams application would result " +
836849
"in an error as it is meant to be used only from Plain consumer client.";
837850

838-
/** {@code window.size.ms} */
851+
/**
852+
* {@code window.size.ms}
853+
*
854+
* @deprecated since 4.1.0.
855+
* Use {@link TimeWindowedDeserializer#WINDOW_SIZE_MS_CONFIG} for {@link TimeWindowedDeserializer}.
856+
*/
857+
@Deprecated
839858
public static final String WINDOW_SIZE_MS_CONFIG = "window.size.ms";
840859
private static final String WINDOW_SIZE_MS_DOC = "Sets window size for the deserializer in order to calculate window end times.";
841860

‎streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedDeserializer.java

Lines changed: 34 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,20 @@
2323
import org.apache.kafka.streams.StreamsConfig;
2424
import org.apache.kafka.streams.state.internals.SessionKeySchema;
2525

26+
import org.slf4j.Logger;
27+
import org.slf4j.LoggerFactory;
28+
2629
import java.util.Map;
2730

2831
public class SessionWindowedDeserializer<T> implements Deserializer<Windowed<T>> {
2932

33+
/**
34+
* Default deserializer for the inner deserializer class of a windowed record. Must implement the {@link Serde} interface.
35+
*/
36+
public static final String WINDOWED_INNER_DESERIALIZER_CLASS = "windowed.inner.deserializer.class";
37+
38+
private final Logger log = LoggerFactory.getLogger(SessionWindowedDeserializer.class);
39+
3040
private Deserializer<T> inner;
3141

3242
// Default constructor needed by Kafka
@@ -36,34 +46,43 @@ public SessionWindowedDeserializer(final Deserializer<T> inner) {
3646
this.inner = inner;
3747
}
3848

39-
@SuppressWarnings("unchecked")
49+
@SuppressWarnings({"deprecation", "unchecked"})
4050
@Override
4151
public void configure(final Map<String, ?> configs, final boolean isKey) {
42-
final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
43-
44-
Serde<T> windowInnerClassSerde = null;
52+
String deserializerConfigKey = WINDOWED_INNER_DESERIALIZER_CLASS;
53+
String deserializerConfigValue = (String) configs.get(WINDOWED_INNER_DESERIALIZER_CLASS);
54+
if (deserializerConfigValue == null) {
55+
final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
56+
if (windowedInnerClassSerdeConfig != null) {
57+
deserializerConfigKey = StreamsConfig.WINDOWED_INNER_CLASS_SERDE;
58+
deserializerConfigValue = windowedInnerClassSerdeConfig;
59+
log.warn("Config {} is deprecated. Please use {} instead.",
60+
StreamsConfig.WINDOWED_INNER_CLASS_SERDE, WINDOWED_INNER_DESERIALIZER_CLASS);
61+
}
62+
}
4563

46-
if (windowedInnerClassSerdeConfig != null) {
64+
Serde<T> windowedInnerDeserializerClass = null;
65+
if (deserializerConfigValue != null) {
4766
try {
48-
windowInnerClassSerde = Utils.newInstance(windowedInnerClassSerdeConfig, Serde.class);
67+
windowedInnerDeserializerClass = Utils.newInstance(deserializerConfigValue, Serde.class);
4968
} catch (final ClassNotFoundException e) {
50-
throw new ConfigException(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, windowedInnerClassSerdeConfig,
51-
"Serde class " + windowedInnerClassSerdeConfig + " could not be found.");
69+
throw new ConfigException(deserializerConfigKey, deserializerConfigValue,
70+
"Serde class " + deserializerConfigValue + " could not be found.");
5271
}
5372
}
5473

55-
if (inner != null && windowedInnerClassSerdeConfig != null) {
56-
if (!inner.getClass().getName().equals(windowInnerClassSerde.deserializer().getClass().getName())) {
74+
if (inner != null && deserializerConfigValue != null) {
75+
if (!inner.getClass().getName().equals(windowedInnerDeserializerClass.deserializer().getClass().getName())) {
5776
throw new IllegalArgumentException("Inner class deserializer set using constructor "
5877
+ "(" + inner.getClass().getName() + ")" +
59-
" is different from the one set in windowed.inner.class.serde config " +
60-
"(" + windowInnerClassSerde.deserializer().getClass().getName() + ").");
78+
" is different from the one set in " + deserializerConfigKey + " config " +
79+
"(" + windowedInnerDeserializerClass.deserializer().getClass().getName() + ").");
6180
}
62-
} else if (inner == null && windowedInnerClassSerdeConfig == null) {
81+
} else if (inner == null && deserializerConfigValue == null) {
6382
throw new IllegalArgumentException("Inner class deserializer should be set either via constructor " +
64-
"or via the windowed.inner.class.serde config");
83+
"or via the " + WINDOWED_INNER_DESERIALIZER_CLASS + " config");
6584
} else if (inner == null)
66-
inner = windowInnerClassSerde.deserializer();
85+
inner = windowedInnerDeserializerClass.deserializer();
6786
}
6887

6988
@Override

‎streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindowedSerializer.java

Lines changed: 34 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,20 @@
2424
import org.apache.kafka.streams.kstream.internals.WindowedSerializer;
2525
import org.apache.kafka.streams.state.internals.SessionKeySchema;
2626

27+
import org.slf4j.Logger;
28+
import org.slf4j.LoggerFactory;
29+
2730
import java.util.Map;
2831

2932
public class SessionWindowedSerializer<T> implements WindowedSerializer<T> {
3033

34+
/**
35+
* Default serializer for the inner serializer class of a windowed record. Must implement the {@link Serde} interface.
36+
*/
37+
public static final String WINDOWED_INNER_SERIALIZER_CLASS = "windowed.inner.serializer.class";
38+
39+
private final Logger log = LoggerFactory.getLogger(SessionWindowedSerializer.class);
40+
3141
private Serializer<T> inner;
3242

3343
// Default constructor needed by Kafka
@@ -37,32 +47,42 @@ public SessionWindowedSerializer(final Serializer<T> inner) {
3747
this.inner = inner;
3848
}
3949

40-
@SuppressWarnings("unchecked")
50+
@SuppressWarnings({"deprecation", "unchecked"})
4151
@Override
4252
public void configure(final Map<String, ?> configs, final boolean isKey) {
43-
final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
44-
Serde<T> windowInnerClassSerde = null;
45-
if (windowedInnerClassSerdeConfig != null) {
53+
String serializerConfigKey = WINDOWED_INNER_SERIALIZER_CLASS;
54+
String serializerConfigValue = (String) configs.get(WINDOWED_INNER_SERIALIZER_CLASS);
55+
if (serializerConfigValue == null) {
56+
final String windowedInnerClassSerdeConfig = (String) configs.get(StreamsConfig.WINDOWED_INNER_CLASS_SERDE);
57+
if (windowedInnerClassSerdeConfig != null) {
58+
serializerConfigKey = StreamsConfig.WINDOWED_INNER_CLASS_SERDE;
59+
serializerConfigValue = windowedInnerClassSerdeConfig;
60+
log.warn("Config {} is deprecated. Please use {} instead.",
61+
StreamsConfig.WINDOWED_INNER_CLASS_SERDE, WINDOWED_INNER_SERIALIZER_CLASS);
62+
}
63+
}
64+
Serde<T> windowedInnerSerializerClass = null;
65+
if (serializerConfigValue != null) {
4666
try {
47-
windowInnerClassSerde = Utils.newInstance(windowedInnerClassSerdeConfig, Serde.class);
67+
windowedInnerSerializerClass = Utils.newInstance(serializerConfigValue, Serde.class);
4868
} catch (final ClassNotFoundException e) {
49-
throw new ConfigException(StreamsConfig.WINDOWED_INNER_CLASS_SERDE, windowedInnerClassSerdeConfig,
50-
"Serde class " + windowedInnerClassSerdeConfig + " could not be found.");
69+
throw new ConfigException(serializerConfigKey, serializerConfigValue,
70+
"Serde class " + serializerConfigValue + " could not be found.");
5171
}
5272
}
5373

54-
if (inner != null && windowedInnerClassSerdeConfig != null) {
55-
if (!inner.getClass().getName().equals(windowInnerClassSerde.serializer().getClass().getName())) {
74+
if (inner != null && serializerConfigValue != null) {
75+
if (!inner.getClass().getName().equals(windowedInnerSerializerClass.serializer().getClass().getName())) {
5676
throw new IllegalArgumentException("Inner class serializer set using constructor "
5777
+ "(" + inner.getClass().getName() + ")" +
58-
" is different from the one set in windowed.inner.class.serde config " +
59-
"(" + windowInnerClassSerde.serializer().getClass().getName() + ").");
78+
" is different from the one set in " + serializerConfigKey + " config " +
79+
"(" + windowedInnerSerializerClass.serializer().getClass().getName() + ").");
6080
}
61-
} else if (inner == null && windowedInnerClassSerdeConfig == null) {
81+
} else if (inner == null && serializerConfigValue == null) {
6282
throw new IllegalArgumentException("Inner class serializer should be set either via constructor " +
63-
"or via the windowed.inner.class.serde config");
83+
"or via the " + WINDOWED_INNER_SERIALIZER_CLASS + " config");
6484
} else if (inner == null)
65-
inner = windowInnerClassSerde.serializer();
85+
inner = windowedInnerSerializerClass.serializer();
6686
}
6787

6888
@Override

0 commit comments

Comments
 (0)