Skip to content

KAFKA-10810: Replace stream threads #9697

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vvcephei merged 10 commits into from
Dec 11, 2020
Merged

KAFKA-10810: Replace stream threads #9697

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
61cdbef
init commit
wcarlson5 Dec 4, 2020
79b743a
global thread case
wcarlson5 Dec 5, 2020
d05ef97
undo
wcarlson5 Dec 5, 2020
0bd4aaf
undo
wcarlson5 Dec 5, 2020
060a0a4
Leah's comments
wcarlson5 Dec 7, 2020
d211da9
John's comments
wcarlson5 Dec 9, 2020
0708ca0
Bruno's comments
wcarlson5 Dec 10, 2020
29fa8c0
Bruno's comments
wcarlson5 Dec 10, 2020
d5b170c
add test for the global thread having an error
wcarlson5 Dec 11, 2020
6e96114
cleanup
wcarlson5 Dec 11, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,6 +436,26 @@ private void defaultStreamsUncaughtExceptionHandler(final Throwable throwable) {
}
}

private void replaceThreadHelper(final Throwable throwable) {
Comment thread
wcarlson5 marked this conversation as resolved.
Outdated
if (globalStreamThread != null && Thread.currentThread().getName().equals(globalStreamThread.getName())) {
log.warn("The global thread cannot be replaced. Reverting to shutting down the client.");
log.error("Encountered the following exception during processing " +
" The streams client is going to shut down now. ", throwable);
closeToError();
}
final StreamThread deadThread = (StreamThread) Thread.currentThread();
threads.remove(deadThread);
addStreamThread();
deadThread.shutdown();
if (throwable instanceof RuntimeException) {
throw (RuntimeException) throwable;
} else if (throwable instanceof Error) {
throw (Error) throwable;
} else {
throw new RuntimeException("Unexpected checked exception caught in the uncaught exception handler", throwable);
}
}

private void handleStreamsUncaughtException(final Throwable throwable,
final StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
final StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse action = streamsUncaughtExceptionHandler.handle(throwable);
Expand All @@ -444,6 +464,9 @@ private void handleStreamsUncaughtException(final Throwable throwable,
"The old handler will be ignored as long as a new handler is set.");
}
switch (action) {
case REPLACE_THREAD:
replaceThreadHelper(throwable);
break;
case SHUTDOWN_CLIENT:
log.error("Encountered the following exception during processing " +
"and the registered exception handler opted to " + action + "." +
Expand Down
1 change: 1 addition & 0 deletions streams/src/main/java/org/apache/kafka/streams/errors/StreamsUncaughtExceptionHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ public interface StreamsUncaughtExceptionHandler {
* Enumeration that describes the response from the exception handler.
*/
enum StreamThreadExceptionResponse {
REPLACE_THREAD(0, "REPLACE_STREAM_THREAD"),

@cadonna cadonna Dec 11, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would rename it to REPLACE_STREAM_THREAD.

@vvcephei vvcephei Dec 11, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, good catch.

Just a quick question: did we misname this option in the KIP? A StreamThread is a specific kind of thread. What I mean is that a GlobalStreamThread is not a StreamThread. Perhaps REPLACE_THREAD and SHUTDOWN_THREAD would have been better, more general names, for these. If you agree, I think we can just amend the KIP and fix it in this PR.

@wcarlson5 wcarlson5 Dec 11, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It should be the string is REPLACE_THREAD. Ill just fix that, its just left over from a period of time where we thought of calling it REPLACE_STREAM_THREAD

@cadonna cadonna Dec 11, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I actually would be in favor of calling the enum value REPLACE_STREAM_THREAD. A stream thread is a stream thread and a global stream thread is a global stream thread. I am aware that the KIP calls the enum value differently, but we also have a config that is called 'NUM_STREAM_THREADS_CONFIG' and we have also 'addStreamThread()' and removeStreamThread(). So I guess, the name to the outside of this is stream thread and not thread. We have also other threads in Kafka Streams like the state directory cleaner thread and the RocksDB metrics recording thread.

@wcarlson5 wcarlson5 Dec 11, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What if we add an option to replace the global thread?

@cadonna cadonna Dec 11, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, now I got it! Sorry! Makes sense! In that case we can reuse REPLACE_THREAD also for the global stream thread. Forgot about that!

SHUTDOWN_CLIENT(1, "SHUTDOWN_KAFKA_STREAMS_CLIENT"),
SHUTDOWN_APPLICATION(2, "SHUTDOWN_KAFKA_STREAMS_APPLICATION");

Expand Down
94 changes: 83 additions & 11 deletions .../org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,12 @@
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Named;
import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.internals.KeyValueStoreBuilder;
import org.apache.kafka.test.IntegrationTest;
import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
Expand All @@ -48,11 +51,13 @@
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.common.utils.Utils.mkObjectProperties;
import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD;
import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.purgeLocalStreamsState;
Expand All @@ -66,7 +71,7 @@
@SuppressWarnings("deprecation") //Need to call the old handler, will remove those calls when the old handler is removed
public class StreamsUncaughtExceptionHandlerIntegrationTest {
@ClassRule
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1, new Properties(), 0L, 0L);
public static final Duration DEFAULT_DURATION = Duration.ofSeconds(30);

@Rule
Expand All @@ -77,6 +82,7 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
private static Properties properties;
private static List<String> processorValueCollector;
private static String appId = "";
private static AtomicBoolean throwError = new AtomicBoolean(true);

@Before
public void setup() {
Expand Down Expand Up @@ -145,6 +151,16 @@ public void shouldShutdownClient() throws InterruptedException {
}
}

@Test
public void shouldReplaceThreads() throws InterruptedException {
testReplaceThreads(2);
}

@Test
public void shouldReplaceSingleThread() throws InterruptedException {
testReplaceThreads(1);
}

@Test
public void shouldShutdownMultipleThreadApplication() throws InterruptedException {
testShutdownApplication(2);
Expand All @@ -155,16 +171,46 @@ public void shouldShutdownSingleThreadApplication() throws InterruptedException
testShutdownApplication(1);
}

@Test
public void testGlobalThreadException() throws InterruptedException {
Comment thread
wcarlson5 marked this conversation as resolved.
Outdated
builder = new StreamsBuilder();
builder.addGlobalStore(
new KeyValueStoreBuilder<>(
Stores.persistentKeyValueStore("globalStore"),
Serdes.String(),
Serdes.String(),
CLUSTER.time
),
inputTopic,
Consumed.with(Serdes.String(), Serdes.String()),
() -> new ShutdownProcessor(processorValueCollector)
);
properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 0);

try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler"));
kafkaStreams.setUncaughtExceptionHandler(exception -> SHUTDOWN_CLIENT);
Comment thread
wcarlson5 marked this conversation as resolved.
Outdated

StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);

produceMessages(0L, inputTopic, "A");
waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.NOT_RUNNING, DEFAULT_DURATION);

assertThat(processorValueCollector.size(), equalTo(1));
}

}

private void produceMessages(final long timestamp, final String streamOneInput, final String msg) {
IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
streamOneInput,
Collections.singletonList(new KeyValue<>("1", msg)),
TestUtils.producerConfig(
CLUSTER.bootstrapServers(),
StringSerializer.class,
StringSerializer.class,
new Properties()),
timestamp);
streamOneInput,
Collections.singletonList(new KeyValue<>("1", msg)),
TestUtils.producerConfig(
CLUSTER.bootstrapServers(),
StringSerializer.class,
StringSerializer.class,
new Properties()),
timestamp);
}

private static class ShutdownProcessor extends AbstractProcessor<String, String> {
Expand All @@ -177,7 +223,10 @@ private static class ShutdownProcessor extends AbstractProcessor<String, String>
@Override
public void process(final String key, final String value) {
valueList.add(value + " " + context.taskId());
throw new StreamsException(Thread.currentThread().getName());
if (throwError.get()) {
throw new StreamsException(Thread.currentThread().getName());
}
throwError.set(true);
}
}

Expand All @@ -202,6 +251,29 @@ private void testShutdownApplication(final int numThreads) throws InterruptedExc
assertThat(processorValueCollector.size(), equalTo(1));
}
}
}

private void testReplaceThreads(final int numThreads) throws InterruptedException {
properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numThreads);
try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler"));

final AtomicInteger count = new AtomicInteger();
kafkaStreams.setUncaughtExceptionHandler(exception -> {
if (count.incrementAndGet() == numThreads) {
throwError.set(false);
}
return REPLACE_THREAD;
});
StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);

produceMessages(0L, inputTopic, "A");
TestUtils.waitForCondition(() -> count.get() == numThreads, "finished replacing threads");
TestUtils.waitForCondition(() -> throwError.get(), "finished replacing threads");
kafkaStreams.close();
waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.NOT_RUNNING, DEFAULT_DURATION);

assertThat("All initial threads have failed and the replacement thread had processed on record",
processorValueCollector.size(), equalTo(numThreads + 1));
}
}
}
Comment thread
wcarlson5 marked this conversation as resolved.