mjsax commented on code in PR #16988:
URL: https://github.com/apache/kafka/pull/16988#discussion_r1757679681
##########
streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java:
##########
@@ -1085,77 +1086,72 @@ public void shouldAllowToQueryAfterThreadDied() throws
Exception {
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> input = builder.stream(streamOne);
input
- .groupByKey()
- .reduce((value1, value2) -> {
- if (value1.length() > 1) {
- if (beforeFailure.compareAndSet(true, false)) {
- throw new RuntimeException("Injected test exception");
+ .groupByKey()
+ .reduce((value1, value2) -> {
+ if (value1.length() > 1) {
+ if (beforeFailure.compareAndSet(true, false)) {
+ throw new RuntimeException("Injected test
exception");
+ }
}
- }
- return value1 + value2;
- }, Materialized.as(storeName))
- .toStream()
- .to(outputTopic);
+ return value1 + value2;
+ }, Materialized.as(storeName))
+ .toStream()
+ .to(outputTopic);
streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2);
kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
- kafkaStreams.setUncaughtExceptionHandler((t, e) -> failed.set(true));
+ kafkaStreams.setUncaughtExceptionHandler(exception -> REPLACE_THREAD);
Review Comment:
Yes, but the test expliclity want to test, that we can still query after the
thread died -- thus, re-starting is something we don't want to do, but we want
the task to migrate to the second/remaining thread.
--
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]
