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 all 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
34 changes: 27 additions & 7 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("Ignoring request to transit 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("Ignoring request to transit 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("Ignoring request to transit 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,17 +274,23 @@ 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.debug(
"Skipping task creation in rebalance because we are already in {} state.",
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.debug(
"Encountered assignment error during partition assignment: {}. Skipping task initialization",
streamThread.assignmentErrorCode
);
} else {
log.debug("Creating tasks based on assignment.");
taskManager.createTasks(assignment);
}
} catch (final Throwable t) {
log.error(
"Error caught during partition assignment, " +
"will abort the current process and re-throw at the end of rebalance: {}",
t
);
"will abort the current process and re-throw at the end of rebalance", t);

@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.

Nice catch.

streamThread.setRebalanceException(t);
} finally {
log.info("partition assignment took {} ms.\n" +
Expand Down Expand Up @@ -809,7 +821,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 +841,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 #onPartitionsAssigned().
// 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.debug("State already transits to {}, skipping the run once call after poll request", state);
return;
}

final long pollLatency = advanceNowAndComputeLatency();

if (records != null && !records.isEmpty()) {
Expand Down
82 changes: 82 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 @@ -20,8 +20,10 @@
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.Cluster;
Expand Down Expand Up @@ -74,6 +76,7 @@
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
Expand Down Expand Up @@ -695,6 +698,85 @@ public void shouldShutdownTaskManagerOnCloseWithoutStart() {
EasyMock.verify(taskManager);
}

@Test
public void shouldNotThrowWhenPendingShutdownInRunOnce() {
mockRunOnce(true);
}

@Test
public void shouldNotThrowWithoutPendingShutdownInRunOnce() {
// A reference test to verify that without intermediate shutdown the runOnce should pass
// without any exception.
mockRunOnce(false);
}

private void mockRunOnce(final boolean shutdownOnPoll) {
final Collection<TopicPartition> assignedPartitions = Collections.singletonList(t1p1);
class MockStreamThreadConsumer<K, V> extends MockConsumer<K, V> {

private StreamThread streamThread;

private MockStreamThreadConsumer(final OffsetResetStrategy offsetResetStrategy) {
super(offsetResetStrategy);
}

@Override
public synchronized ConsumerRecords<K, V> poll(final Duration timeout) {
assertNotNull(streamThread);
if (shutdownOnPoll) {
streamThread.shutdown();
Comment thread
abbccdda marked this conversation as resolved.
}
streamThread.rebalanceListener.onPartitionsAssigned(assignedPartitions);
return super.poll(timeout);
}

private void setStreamThread(final StreamThread streamThread) {
this.streamThread = streamThread;
}
}

final MockStreamThreadConsumer<byte[], byte[]> mockStreamThreadConsumer =
new MockStreamThreadConsumer<>(OffsetResetStrategy.EARLIEST);

final TaskManager taskManager = new TaskManager(new MockChangelogReader(),
processId,
"log-prefix",
mockStreamThreadConsumer,
streamsMetadataState,
null,
null,
null,
new AssignedStreamsTasks(new LogContext()),
new AssignedStandbyTasks(new LogContext()));
taskManager.setConsumer(mockStreamThreadConsumer);
taskManager.setAssignmentMetadata(Collections.emptyMap(), Collections.emptyMap());

final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(metrics, clientId);
final StreamThread thread = new StreamThread(
mockTime,
config,
null,
mockStreamThreadConsumer,
mockStreamThreadConsumer,
null,
taskManager,
streamsMetrics,
internalTopologyBuilder,
clientId,
new LogContext(""),
new AtomicInteger()
).updateThreadMetadata(getSharedAdminClientId(clientId));

mockStreamThreadConsumer.setStreamThread(thread);
mockStreamThreadConsumer.assign(assignedPartitions);
mockStreamThreadConsumer.updateBeginningOffsets(Collections.singletonMap(t1p1, 0L));

addRecord(mockStreamThreadConsumer, 1L, 0L);
thread.setState(StreamThread.State.STARTING);
thread.setState(StreamThread.State.PARTITIONS_REVOKED);
thread.runOnce();
}

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