KAFKA-10810: Replace stream threads by wcarlson5 · Pull Request #9697 · apache/kafka

wcarlson5 · 2020-12-04T22:17:21Z

StreamThreads can now be replaced in the streams uncaught exception handler

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

wcarlson5

@cadonna part 6 is ready for review

wcarlson5 · 2020-12-04T22:19:21Z

We need to throw the error or the task gets lost and we drop records

wcarlson5 · 2020-12-04T22:20:41Z

removing from the thread list does 2 things

keeps DEAD threads out of the list as kip-663 dictates

ensures the next thread has the same name

To ensure #2 holds, do we need to have a mutex on thread creation? Or does it not matter?

It won't matter. Ensures is too strong of a word. IF there has been other threads removed before this it may have a different name. However this ensures that the replacement thread will never have a thread index larger than the number of threads

vvcephei · 2020-12-09T19:25:55Z

+                            " The streams client is going to shut down now. ", throwable);
+                    close(Duration.ZERO);
+                }
+                final StreamThread deadThread = (StreamThread) threads.stream().filter(n -> n.getName().equals(Thread.currentThread().getName())).toArray()[0];


Is this equivalent to Thread.currentThead()?

Yep, it is now :)

vvcephei · 2020-12-09T20:16:13Z

+            produceMessages(0L, inputTopic, "A");
+            waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.NOT_RUNNING, DEFAULT_DURATION);
+
+            assertThat(processorValueCollector.size(), equalTo(3));


It's not obvious to me how this verifies that the thread actually got replaced. Maybe an explanatory comment is in order?

vvcephei

Thanks, @wcarlson5 ! I had a few thoughts above.

cadonna

Thanks for the PR @wcarlson5 !

Here my feedback!

cadonna · 2020-12-10T15:12:51Z

+                final StreamThread deadThread = (StreamThread) Thread.currentThread();
+                threads.remove(deadThread);
+                addStreamThread();
+                deadThread.shutdown();


Do we need to shutdown the dead stream thread? completeShutDown() will be called anyways.

I don't think it matters, it just set the thread state earlier, but we can delete it

cadonna · 2020-12-10T15:14:15Z

+            case REPLACE_THREAD:
+                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 " +
+                            "and the registered exception handler opted to " + action + "." +
+                            " The streams client is going to shut down now. ", throwable);
+                    close(Duration.ZERO);
+                }
+                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);
+                }


I think it would be cleaner to extract this code to a separate method.

we can do that

cadonna · 2020-12-10T15:46:26Z

+            waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.NOT_RUNNING, DEFAULT_DURATION);
+
+            assertThat(processorValueCollector.size(), equalTo(3));
+            //because we only have 2 threads at the start and each record kills a thread we must have replaced threads


Could you please be a bit clearer in the explanatory comment? BTW, we execute this test also once with just one stream thread so the 2 stream threads in the comment are not correct. Also, wouldn't it be better to explain the verification in the call to assertThat() instead of in a comment? You can pass a reason to the method.

I edited the test to be based on the number of thread instead of hard coding. And gave a reason

cadonna · 2020-12-10T15:53:58Z

+                count.getAndIncrement();
+                if (count.get() > 2) {
+                    return SHUTDOWN_CLIENT;
+                }
+                return REPLACE_THREAD;
+            });


I think it would be better to have a test that shows that a new thread that replaced a failed one, actually is able to process records. So, I would let the new thread process some records and then shutdown the client with a normal close.

Maybe similar applies to the shutdown tests. First let the client/application process some records and then throw an exception that shuts down the client/application. I guess, this last paragraph is something for a separate PR.

We can change the test so that we verify the replaced threads can process records.

I am not sure that is necessary for the shutdown as testing if streams can process some records once started should be tested elsewhere, but in any case I think that the PR is not the place for this discussion

wcarlson5

@cadonna here are some changes

wcarlson5 · 2020-12-10T16:25:54Z

+                final StreamThread deadThread = (StreamThread) Thread.currentThread();
+                threads.remove(deadThread);
+                addStreamThread();
+                deadThread.shutdown();


I don't think it matters, it just set the thread state earlier, but we can delete it

wcarlson5 · 2020-12-10T16:28:26Z

+            case REPLACE_THREAD:
+                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 " +
+                            "and the registered exception handler opted to " + action + "." +
+                            " The streams client is going to shut down now. ", throwable);
+                    close(Duration.ZERO);
+                }
+                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);
+                }


we can do that

wcarlson5 · 2020-12-10T16:37:00Z

+            waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.NOT_RUNNING, DEFAULT_DURATION);
+
+            assertThat(processorValueCollector.size(), equalTo(3));
+            //because we only have 2 threads at the start and each record kills a thread we must have replaced threads


I edited the test to be based on the number of thread instead of hard coding. And gave a reason

wcarlson5 · 2020-12-10T16:52:03Z

+                count.getAndIncrement();
+                if (count.get() > 2) {
+                    return SHUTDOWN_CLIENT;
+                }
+                return REPLACE_THREAD;
+            });


We can change the test so that we verify the replaced threads can process records.

I am not sure that is necessary for the shutdown as testing if streams can process some records once started should be tested elsewhere, but in any case I think that the PR is not the place for this discussion

cadonna · 2020-12-11T11:06:38Z

     * Enumeration that describes the response from the exception handler.
     */
    enum StreamThreadExceptionResponse {
+        REPLACE_THREAD(0, "REPLACE_STREAM_THREAD"),


I would rename it to REPLACE_STREAM_THREAD.

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.

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

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.

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

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!

cadonna

Thanks @wcarlson5!

LGTM!

vvcephei

Thanks @wcarlson5 !

zydzjy · 2021-01-08T08:14:09Z

what can i do with replace new thread?Can i skip the current offset in exception handler,cause the exception occured by the data with current offset?

wcarlson5 · 2021-01-11T16:28:30Z

@zydzjy I don't think you can do that yet, at least not with anything I have added here. The idea is to recover from transient expectations without having to restart the client.

wcarlson5 commented Dec 4, 2020

View reviewed changes

lct45 reviewed Dec 7, 2020

View reviewed changes

Comment thread streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java Outdated

wcarlson5 added 5 commits December 9, 2020 08:57

init commit

61cdbef

global thread case

79b743a

undo

d05ef97

undo

0bd4aaf

Leah's comments

060a0a4

vvcephei reviewed Dec 9, 2020

View reviewed changes

John's comments

d211da9

cadonna reviewed Dec 10, 2020

View reviewed changes

Bruno's comments

0708ca0

wcarlson5 commented Dec 10, 2020

View reviewed changes

wcarlson5 added 2 commits December 10, 2020 10:49

Bruno's comments

29fa8c0

add test for the global thread having an error

d5b170c

cadonna reviewed Dec 11, 2020

View reviewed changes

cleanup

6e96114

cadonna approved these changes Dec 11, 2020

View reviewed changes

vvcephei approved these changes Dec 11, 2020

View reviewed changes

vvcephei added the streams label Dec 11, 2020

vvcephei merged commit d5dc7df into apache:trunk Dec 11, 2020

wcarlson5 deleted the wcarlson-replace-thread branch December 11, 2020 21:09

mjsax added the kip Requires or implements a KIP label Jan 8, 2021

Conversation

wcarlson5 commented Dec 4, 2020

Committer Checklist (excluded from commit message)

Uh oh!

wcarlson5 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wcarlson5 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

zydzjy commented Jan 8, 2021