aliehsaeedii commented on code in PR #21754:
URL: https://github.com/apache/kafka/pull/21754#discussion_r3557289666
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -121,6 +123,56 @@ private Properties props() {
return streamsConfiguration;
}
+ @SuppressWarnings("rawtypes")
+ private void buildAndStart(final StoreBuilder storeBuilder,
+ final ProcessorSupplier<String, String, Void,
Void> processorSupplier,
+ final String storeName,
+ final Properties props) throws Exception {
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.addStateStore(storeBuilder)
+ .stream(inputStream, Consumed.with(Serdes.String(),
Serdes.String()))
+ .process(processorSupplier, storeName);
+ kafkaStreams = new KafkaStreams(builder.build(), props);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
+ }
+
+ @SuppressWarnings("rawtypes")
+ private void assertDowngradeThrowsProcessorStateException(
+ final StoreBuilder storeBuilder,
+ final ProcessorSupplier<String, String, Void, Void>
processorSupplier,
+ final String storeName,
+ final Properties props) {
+ boolean exceptionThrown = false;
+ try {
+ buildAndStart(storeBuilder, processorSupplier, storeName, props);
+ } catch (final Exception e) {
+ Throwable cause = e;
+ while (cause != null) {
+ if (cause instanceof ProcessorStateException &&
+ cause.getMessage() != null &&
+ cause.getMessage().contains("headers-aware") &&
+ cause.getMessage().contains("Downgrade")) {
+ exceptionThrown = true;
+ break;
+ }
+ cause = cause.getCause();
+ }
+ if (!exceptionThrown) {
+ throw new AssertionError(
+ "Expected ProcessorStateException about downgrade not
being supported, but got: " + e.getMessage(), e);
+ }
+ } finally {
+ if (kafkaStreams != null) {
+ kafkaStreams.close(Duration.ofSeconds(30L));
+ kafkaStreams = null;
Review Comment:
Nulling `kafkaStreams` here makes `@AfterEach shutdown()` skip `cleanUp()`
for the fail-downgrade tests. The old inline code left the field non-null, so
`cleanUp()` still ran. It's harmless (each test has its own app id and temp
dir), but it's a behavior change from a refactor meant to preserve behavior —
drop the `= null` if you want `cleanUp()` to keep running.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -121,6 +123,56 @@ private Properties props() {
return streamsConfiguration;
}
+ @SuppressWarnings("rawtypes")
+ private void buildAndStart(final StoreBuilder storeBuilder,
Review Comment:
`addStateStore` takes `StoreBuilder<?>`, so you can type this param as
`StoreBuilder<?>` and drop the `@SuppressWarnings("rawtypes")`. Same for
`assertDowngradeThrowsProcessorStateException`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
