muralibasani commented on code in PR #21754:
URL: https://github.com/apache/kafka/pull/21754#discussion_r3581597852
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -121,6 +123,53 @@ private Properties props() {
return streamsConfiguration;
}
+ 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);
+ }
+
+ 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));
+ }
+ }
+ if (!exceptionThrown) {
+ throw new AssertionError(
+ "Expected ProcessorStateException to be thrown when attempting
to downgrade from headers-aware store");
Review Comment:
Good one, fixed. Added a description param in to all of them.
--
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]
