muralibasani commented on code in PR #21754:
URL: https://github.com/apache/kafka/pull/21754#discussion_r3573896743
##########
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:
Good one, updated.
--
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]
