wcarlson5 commented on a change in pull request #9629:
URL: https://github.com/apache/kafka/pull/9629#discussion_r528992986
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
##########
@@ -110,83 +110,45 @@ public void teardown() throws IOException {
}
@Test
- public void shouldShutdownThreadUsingOldHandler() throws Exception {
+ public void shouldShutdownThreadUsingOldHandler() throws
InterruptedException {
try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
- final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean flag = new AtomicBoolean(false);
kafkaStreams.setUncaughtExceptionHandler((t, e) -> flag.set(true));
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
-
produceMessages(0L, inputTopic, "A");
- waitForApplicationState(Collections.singletonList(kafkaStreams),
KafkaStreams.State.ERROR, Duration.ofSeconds(15));
TestUtils.waitForCondition(flag::get, "Handler was called");
- assertThat(processorValueCollector.size(), equalTo(2));
Review comment:
as above
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
##########
@@ -97,7 +97,7 @@ public void setup() {
mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers()),
mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId),
mkEntry(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getPath()),
- mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2),
+ mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1),
Review comment:
Yes, Both the old handler test and the close client should have 2
threads. We need to ensure that after a rebalance the old handler has attempted
the process the record twice and the client shutdown only once. We can not be
sure of that with only one thread.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
##########
@@ -110,83 +110,45 @@ public void teardown() throws IOException {
}
@Test
- public void shouldShutdownThreadUsingOldHandler() throws Exception {
+ public void shouldShutdownThreadUsingOldHandler() throws
InterruptedException {
try (final KafkaStreams kafkaStreams = new
KafkaStreams(builder.build(), properties)) {
- final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean flag = new AtomicBoolean(false);
kafkaStreams.setUncaughtExceptionHandler((t, e) -> flag.set(true));
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
-
produceMessages(0L, inputTopic, "A");
- waitForApplicationState(Collections.singletonList(kafkaStreams),
KafkaStreams.State.ERROR, Duration.ofSeconds(15));
TestUtils.waitForCondition(flag::get, "Handler was called");
+ waitForApplicationState(Collections.singletonList(kafkaStreams),
KafkaStreams.State.ERROR, DEFAULT_DURATION);
Review comment:
The order is not really that important here, either way works
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
