Skip to content

KAFKA-9231: Streams Threads may die from recoverable errors with EOS enabled #7748

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 4, 2019
Merged

KAFKA-9231: Streams Threads may die from recoverable errors with EOS enabled #7748

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
bd0bc94
KAFKA-9231: Streams: Rebalance when fenced in suspend
vvcephei Nov 26, 2019
19f22a9
sequence catch after finally
vvcephei Nov 26, 2019
f6ac2c6
catch retriable API exceptions in flush more systematically
vvcephei Nov 26, 2019
7a7d1e9
catch the right exception
vvcephei Nov 26, 2019
33c36a1
swallow exceptions while closing zombie tasks
vvcephei Nov 28, 2019
02fdedd
cr review
vvcephei Nov 28, 2019
db878d7
trying out a more general approach
vvcephei Nov 29, 2019
e6f7cab
remove unnecessary log
vvcephei Nov 29, 2019
a264c8e
set up RocksDB metrics only after we successfully open the database
vvcephei Nov 29, 2019
96a2d96
add test coverage
vvcephei Dec 2, 2019
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
11 changes: 10 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,13 @@

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.LockException;
import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskMigratedException;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
Expand Down Expand Up @@ -172,7 +174,14 @@ public String toString(final String indent) {
* Flush all state stores owned by this task
*/
void flushState() {
stateMgr.flush();
try {
stateMgr.flush();
} catch (final ProcessorStateException e) {

@mjsax mjsax Nov 27, 2019 •
edited
Loading

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.

Why do we catch it at this level? Seems we should move it into ProcessorStateManger#flush() ?

Also note, that ProcessorStateManger#flush() only throws the first exception what might not be what we want to detect this case correctly.

@vvcephei vvcephei Nov 28, 2019

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.

Yeah, I wasn't sure... This way is a bit roundabout. On the other hand, right now, TaskMigratedException is only thrown by the tasks themselves or the threads that run them. It didn't seem within the responsibility of the ProcessorStateManager to reason about what specific exceptions imply about the state of the task.

@mjsax mjsax Nov 28, 2019

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.

Fair enough. It's not ideal that we get a ProcessorStateException and have to call getCause(), but well, it is what it is.

@guozhangwang guozhangwang Dec 3, 2019

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.

Reading the code here: we throw TaskMigratedException in both StreamTask as well as StreamThread. For the first case, the exception is thrown all the way to the caller of TaskManager and rethrown (hence in the level of StreamThread); the latter case is only a few exceptional situations where the tasks are closed due to underlying exceptions.

I feel throwing the exception in the level of StreamTask is fine, but we did too many capture / rethrow in different layers on top of it, which could be cleaned up in a separate tech-debt PR.

@vvcephei vvcephei Dec 4, 2019

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.

I very much agree. It's quite difficult to trace which exceptions would exist in different parts of the stack, and it would be nice to clean it up in the future.

@guozhangwang guozhangwang Dec 16, 2019

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.

@vvcephei while working on the cleanup PR I realized that when we call store.flush internally we should not blindly wrap all thrown exceptions from stores, but only those not StreamsException -- i.e. if the error is not from the store itself (like an IOException) but from the streams library, then just throw it out directly.

Will do this small refactoring in my PR.

if (e.getCause() instanceof RetriableException) {

@bbejeck bbejeck Nov 26, 2019

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.

Is this intentional? ProducerFencedException doesn't inherit from RetriableException, so unless I'm missing something, this will still kill the thread. Same for below as well in StandbyTask.

@vvcephei vvcephei Nov 26, 2019

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.

Oh, boy. You're right... Really off my game today.

log.warn("Caught a retriable Kafka exception while flushing state. Initiating a rebalance to attempt recovery.", e);
throw new TaskMigratedException(this, e);
}
}
}

/**
Expand Down
3 changes: 3 additions & 0 deletions ...ams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,9 @@ public void update(final ConsumerRecord<byte[], byte[]> record) {
}

public void flushState() {
// this could theoretically throw a ProcessorStateException caused by a kafka.RetriableException,
// but in practice this shouldn't happen for global state update tasks, since the stores are not
// logged and there are no downstream operators after global stores.
stateMgr.flush();
stateMgr.checkpoint(offsets);
}
Expand Down
11 changes: 10 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,13 @@
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.errors.TaskMigratedException;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;

Expand Down Expand Up @@ -122,7 +124,14 @@ public void commit() {
}

private void flushAndCheckpointState() {
stateMgr.flush();
try {
stateMgr.flush();
} catch (final ProcessorStateException e) {
if (e.getCause() instanceof RetriableException) {
log.warn("Caught a retriable Kafka exception while flushing state. Initiating a rebalance to attempt recovery.", e);
throw new TaskMigratedException(this, e);
}
}
stateMgr.checkpoint(Collections.emptyMap());
}

Expand Down