Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

KAFKA-16331: Remove EOSv1 from Kafka Streams integration tests #17110

Merged
merged 1 commit into from
Sep 16, 2024
Merged

KAFKA-16331: Remove EOSv1 from Kafka Streams integration tests #17110

merged 1 commit into from
Sep 16, 2024

Conversation

mjsax
Copy link
Member

@mjsax mjsax commented Sep 6, 2024

We will remove EOSv1 in 4.0, and thus can remove the corresponding integration tests.

@mjsax mjsax added streams tests Test fixes (including flaky tests) labels Sep 6, 2024
mjsax
mjsax commented Sep 6, 2024
View reviewed changes
...ms/src/test/java/org/apache/kafka/streams/integration/EOSUncleanShutdownIntegrationTest.java
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Side improvement

streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
@@ -176,13 +176,13 @@ public void createTopics() throws Exception {
}

@ParameterizedTest
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we run this "eos test" with at-least-once ?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems the at-least-once was added by #11283

Maybe we can have a follow-up to rename this class or move the at-least-once tests out of this class.

Copy link
Member Author

@mjsax mjsax Sep 11, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems this is a mess -- the original PR actually also added "early returns" for some test... Guess we need to do more follow up cleanup.

Seem one root cause if, that we have a centralized parameter originally, but now a per-test one.

But should be its own PR to do this cleanup, after this one was merged.

streams/src/test/java/org/apache/kafka/streams/integration/GlobalKTableEOSIntegrationTest.java
replicatedStore.all().forEachRemaining(pair -> globalState.put(pair.key, pair.value));
try (final KeyValueIterator<Long, String> it = replicatedStore.all()) {
it.forEachRemaining(pair -> globalState.put(pair.key, pair.value));
}
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Side improvement (more of the same below)

...test/java/org/apache/kafka/streams/integration/JoinGracePeriodDurabilityIntegrationTest.java
@ParameterizedTest
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we running this test parameterized?

.../test/java/org/apache/kafka/streams/integration/KTableSourceTopicRestartIntegrationTest.java
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Side improvement

...ms/src/test/java/org/apache/kafka/streams/integration/ResetPartitionTimeIntegrationTest.java
@ParameterizedTest
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we running this test parameterized?

streams/src/test/java/org/apache/kafka/streams/integration/RocksDBMetricsIntegrationTest.java
@@ -145,7 +144,7 @@ private interface MetricsVerifier {
}

@ParameterizedTest
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we running this test parameterized

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree we probably don't need a parameterized test here and on the ResetPartitionTimeIntegrationTest above you called out, but maybe we can address in a followup PR

streams/src/test/java/org/apache/kafka/streams/integration/RocksDBMetricsIntegrationTest.java
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class);
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Side improvement

streams/src/test/java/org/apache/kafka/streams/integration/RocksDBMetricsIntegrationTest.java
streamsConfiguration.put(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG, Sensor.RecordingLevel.DEBUG.name);
streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, processingGuarantee);
streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
streamsConfiguration.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
return streamsConfiguration;
}

@SuppressWarnings("deprecation")
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We are using deprecated TimeWindows.of() below so need this -- should be cleanup up soon with the PR which removes this deprecated method.

@mjsax mjsax mentioned this pull request Sep 11, 2024
Merged
bbejeck
bbejeck approved these changes Sep 16, 2024
View reviewed changes
Copy link
Contributor

@bbejeck bbejeck left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for cleanup @mjsax - LGTM

streams/src/test/java/org/apache/kafka/streams/integration/RocksDBMetricsIntegrationTest.java
@@ -145,7 +144,7 @@ private interface MetricsVerifier {
}

@ParameterizedTest
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
@ValueSource(strings = {StreamsConfig.AT_LEAST_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree we probably don't need a parameterized test here and on the ResetPartitionTimeIntegrationTest above you called out, but maybe we can address in a followup PR

@chia7712 chia7712 merged commit 21e67b3 into apache:trunk Sep 16, 2024
5 of 6 checks passed
@mjsax mjsax deleted the kafka-16331-eosv1-integration-tests branch September 16, 2024 18:17
@mjsax
Copy link
Member Author

mjsax commented Sep 16, 2024

Thanks! Filed https://issues.apache.org/jira/browse/KAFKA-17558

@zzzk1 zzzk1 mentioned this pull request Sep 29, 2024
Merged
3 tasks
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@chia7712 chia7712 chia7712 approved these changes

@bbejeck bbejeck bbejeck approved these changes

Assignees
No one assigned
Labels
streams tests Test fixes (including flaky tests)
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
3 participants
@mjsax @bbejeck @chia7712