Skip to content

KAFKA-8620: fix NPE due to race condition during shutdown while rebalancing #7021

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
mjsax merged 10 commits into from
Jul 15, 2019
Merged

KAFKA-8620: fix NPE due to race condition during shutdown while rebalancing #7021

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
c6a5997
fix the state after poll request
Jul 3, 2019
1337616
add unit test
Jul 3, 2019
d5b7b5c
trying to find a way to pass assignment in
Jul 4, 2019
b2326b4
minor fix
Jul 7, 2019
58b7ce3
address comments
Jul 9, 2019
9c13d42
success reproduce
Jul 9, 2019
f1b1002
address minor suggestion
Jul 9, 2019
eb490a9
avoid info exposure of internal implementation
Jul 9, 2019
a99606b
address comments by initializing real task manager and one-time mock …
Jul 10, 2019
5c6ec62
address Bruno's comment
Jul 10, 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
6 changes: 6 additions & 0 deletions clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,12 @@ public synchronized void assign(Collection<TopicPartition> partitions) {
this.subscriptions.assignFromUser(new HashSet<>(partitions));
}

public synchronized void assignForSubscribed(Collection<TopicPartition> partitions) {
ensureNotClosed();
committed.clear();
this.subscriptions.assignFromSubscribed(new HashSet<>(partitions));
}

@Override
public synchronized void unsubscribe() {
ensureNotClosed();
Expand Down
28 changes: 23 additions & 5 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,14 +190,20 @@ State setState(final State newState) {
oldState = state;

if (state == State.PENDING_SHUTDOWN && newState != State.DEAD) {
log.debug("Invalid transition from PENDING_SHUTDOWN to {}: " +
"only DEAD state is a valid next state", newState);
Comment thread
abbccdda marked this conversation as resolved.
// when the state is already in PENDING_SHUTDOWN, all other transitions will be
// refused but we do not throw exception here
return null;
} else if (state == State.DEAD) {
log.debug("Invalid transition from DEAD to {}: " +
"no valid next state after DEAD", newState);
Comment thread
abbccdda marked this conversation as resolved.
// when the state is already in NOT_RUNNING, all its transitions
// will be refused but we do not throw exception here
return null;
} else if (state == State.PARTITIONS_REVOKED && newState == State.PARTITIONS_REVOKED) {
log.debug("Invalid transition from PARTITIONS_REVOKED to PARTITIONS_REVOKED: " +
"self transition is not allowed");

@mjsax mjsax Jul 9, 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.

Thinking about this, I am wondering if we should just change the FSM to allow this transition and simplify the code here? \cc @guozhangwang

@guozhangwang guozhangwang Jul 15, 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.

I thought about this when tightening the FSM before but the unit tests reminds me of one thing: our current contract is that we only transit to PARTITIONS_REVOKED when calling onPartitionsRevoked, which is called only once at the beginning of the rebalance today, so keeping it strict is better just in case we have incorrect partial rebalance procedure.

With KIP-429 this may be violated so we need to revisit our FSM once Streams adopt cooperative protocols. cc @ableegoldman who's working on this.

// when the state is already in PARTITIONS_REVOKED, its transition to itself will be
// refused but we do not throw exception here
return null;
Expand Down Expand Up @@ -268,9 +274,13 @@ public void onPartitionsAssigned(final Collection<TopicPartition> assignment) {
final long start = time.milliseconds();
try {
if (streamThread.setState(State.PARTITIONS_ASSIGNED) == null) {
return;
Comment thread
abbccdda marked this conversation as resolved.
}
if (streamThread.assignmentErrorCode.get() == StreamsPartitionAssignor.Error.NONE.code()) {
log.error("State transition from {} to PARTITIONS_ASSIGNED failed. " +

@mjsax mjsax Jul 9, 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.

Why is this an ERROR? If we receive a SHUTDOWN signal, it's just an "eager exit" to not create tasks IMHO. We might want to log a DEBUG though.

Would also update the message:

log.debug("Skipping task creation in rebalance because we are already in shutdown phase.");

@abbccdda abbccdda Jul 9, 2019

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Sounds good to me.

"Will skip the task initialization", streamThread.state());
} else if (streamThread.assignmentErrorCode.get() != StreamsPartitionAssignor.Error.NONE.code()) {

@cadonna cadonna Jul 9, 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.

Isn't this a rather brittle fix? How does this code guarantee that after streamThread.assignmentErrorCode.get() != StreamsPartitionAssignor.Error.NONE.code() is false, the result of streamThread.assignmentErrorCode.get() does not change again before the tasks are created?

@abbccdda abbccdda Jul 9, 2019

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

in PartitionsAssigned we should have already passed the assignment error placement, so I won't expect another flip for its value during this round rebalance.

@cadonna cadonna Jul 9, 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.

Makes sense

log.error("Encountered assignment error during partition assignment: {}. " +
"Will skip the task initialization", streamThread.assignmentErrorCode);
Comment thread
abbccdda marked this conversation as resolved.
Outdated

@cadonna cadonna Jul 10, 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.

Shouldn't the indentation be as follows?

log.debug(
    "Encountered assignment error during partition assignment: {}. Will skip the task initialization", 
    streamThread.assignmentErrorCode
);

The first parameter is one or two characters too long, though.

} else {
log.debug("Creating tasks based on assignment.");
taskManager.createTasks(assignment);
}
} catch (final Throwable t) {
Expand Down Expand Up @@ -809,7 +819,6 @@ private void enforceRebalance() {
// Visible for testing
void runOnce() {
final ConsumerRecords<byte[], byte[]> records;

now = time.milliseconds();

if (state == State.PARTITIONS_ASSIGNED) {
Expand All @@ -830,6 +839,15 @@ void runOnce() {
throw new StreamsException(logPrefix + "Unexpected state " + state + " during normal iteration");
}

// Shutdown hook could potentially be triggered and transit the thread state to PENDING_SHUTDOWN during #pollRequests().
// The task manager internal states could be uninitialized if the state transition happens during #onPartitionAssigned().
// Should only proceed when the thread is still running after #pollRequests(), because no external state mutation
// could affect the task manager state beyond this point within #runOnce().
if (!isRunning()) {
log.warn("State already transits to {}, skipping the run once call after poll request", state);

@mjsax mjsax Jul 9, 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.

Seems like this should be DEBUG level?

@abbccdda abbccdda Jul 9, 2019

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

Sounds good

return;
}

final long pollLatency = advanceNowAndComputeLatency();

if (records != null && !records.isEmpty()) {
Expand Down Expand Up @@ -1179,7 +1197,7 @@ private void completeShutdown(final boolean cleanRun) {
// intentionally do not check the returned flag
setState(State.PENDING_SHUTDOWN);

log.info("Shutting down");
log.info("Shutting down with cleanRun {}", cleanRun);

@mjsax mjsax Jul 9, 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.

Not sure if we need this? If the shutdown is not clean, we logged an ERROR before in StreamThread#run()

@cadonna cadonna Jul 9, 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.

I would not use variable names in log messages.

@abbccdda abbccdda Jul 9, 2019

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

I could avoid using variable name in the log, but still feel needed to log the flag here, as we can't control future changes to set cleanRun flag.


try {
taskManager.shutdown(cleanRun);
Expand Down
52 changes: 52 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -695,6 +695,58 @@ public void shouldShutdownTaskManagerOnCloseWithoutStart() {
EasyMock.verify(taskManager);
}

@Test
public void shouldNotAccessTaskManagerWhenPendingShutdownInRunOnce() {

@cadonna cadonna Jul 10, 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.

I would give this test a more meaningful name, e.g. shouldNotThrowWhenPendingShutdownInRunOnce.

final LogCaptureAppender appender = LogCaptureAppender.createAndRegister();

internalTopologyBuilder.addSource(null, "source", null, null, null, topic1);

final StreamThread streamThread = createStreamThread(clientId, new StreamsConfig(configProps(true)), true);

final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
final List<TopicPartition> assignedPartitions = new ArrayList<>();

// assign single partition
assignedPartitions.add(t1p1);
assignedPartitions.add(t1p2);

@mjsax mjsax Jul 9, 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.

Why do we need two partitions?

activeTasks.put(task1, Collections.singleton(t1p1));
activeTasks.put(task2, Collections.singleton(t1p2));

streamThread.taskManager().setAssignmentMetadata(activeTasks, Collections.emptyMap());
final MockConsumer<byte[], byte[]> mockConsumer = (MockConsumer<byte[], byte[]>) streamThread.consumer;
final Map<TopicPartition, Long> beginOffsets = new HashMap<>();
beginOffsets.put(t1p1, 0L);
beginOffsets.put(t1p2, 0L);
mockConsumer.updateBeginningOffsets(beginOffsets);

final Thread callbackThread = new Thread(() -> {

@mjsax mjsax Jul 9, 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.

Why do we need a Thread here?

@abbccdda abbccdda Jul 9, 2019

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

We need a separate thread to trigger callbacks, otherwise the state won't proceed. Maybe there is a better way to do that?

@mjsax mjsax Jul 9, 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.

Can you elaborate? What do you mean by

otherwise the state won't proceed

streamThread.rebalanceListener.onPartitionsRevoked(Collections.emptyList());
mockConsumer.assignForSubscribed(assignedPartitions);
streamThread.rebalanceListener.onPartitionsAssigned(assignedPartitions);
});

streamThread.setStateListener(
(t, newState, oldState) -> {
if (oldState == StreamThread.State.CREATED && newState == StreamThread.State.STARTING) {
// Start triggering consumer callbacks to make sure we proceed to PARTITIONS_ASSIGNED stage.
callbackThread.start();
} else if (oldState == StreamThread.State.PARTITIONS_REVOKED && newState == StreamThread.State.PARTITIONS_ASSIGNED) {
// Immediately trigger shutdown call to switch the state.
streamThread.shutdown();
Comment thread
abbccdda marked this conversation as resolved.
}
});

streamThread.run();

LogCaptureAppender.unregister(appender);
final List<String> strings = appender.getMessages();
assertTrue(strings.contains(
Comment thread
abbccdda marked this conversation as resolved.
Outdated
"stream-thread [clientId-StreamThread-1] " +
"State already transits to PENDING_SHUTDOWN, " +
"skipping the run once call after poll request")
);
}

@Test
public void shouldOnlyShutdownOnce() {
final Consumer<byte[], byte[]> consumer = EasyMock.createNiceMock(Consumer.class);
Expand Down