Skip to content

KAFKA-10150: task state transitions/management and committing cleanup #8856

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
guozhangwang merged 14 commits into from
Jun 16, 2020
Merged

KAFKA-10150: task state transitions/management and committing cleanup #8856

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
3789a7c
enforce SUSPENDED and check commitNeeded
ableegoldman Jun 11, 2020
f621c3a
check if commit needed in TM
ableegoldman Jun 11, 2020
0c3c080
just suspend in closeandRecycleState
ableegoldman Jun 11, 2020
a680583
clean up handleRevocation and handleAssignment
ableegoldman Jun 11, 2020
fb64b97
Task and TaskManager tests
ableegoldman Jun 11, 2020
7b08e28
always close all tasks during shutdown
ableegoldman Jun 12, 2020
dd03646
only commit all for eos-beta
ableegoldman Jun 12, 2020
1f989fc
addressing CR feedback
ableegoldman Jun 13, 2020
cf21fb5
get rid of checkpoint member
ableegoldman Jun 15, 2020
d2ba0df
fixing up last few TM tests
ableegoldman Jun 15, 2020
bd032c9
log before transition
ableegoldman Jun 16, 2020
0349663
attempt to postCommit all tasks
ableegoldman Jun 16, 2020
3ff79ea
always commit when commitNeeded
ableegoldman Jun 16, 2020
2529fb9
only commit active tasks
ableegoldman Jun 16, 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
31 changes: 21 additions & 10 deletions 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 @@ -111,11 +111,25 @@ public void completeRestoration() {

@Override
public void suspend() {
log.trace("No-op suspend with state {}", state());
if (state() == State.RUNNING) {
transitionTo(State.SUSPENDED);
} else if (state() == State.RESTORING) {
throw new IllegalStateException("Illegal state " + state() + " while suspending standby task " + id);
switch (state()) {
case CREATED:
case RUNNING:
log.info("Suspended {}", state());
transitionTo(State.SUSPENDED);

break;

case SUSPENDED:
log.info("Skip suspending since state is {}", state());

break;

case RESTORING:

@vvcephei vvcephei Jun 12, 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.

I get that standbys should never really be in RESTORING state, but it still doesn't seem like it's philosophically any more illegal to suspend from RESTORING than it is from RUNNING. I'd vote to legalize RESTORING here. It does seem like a useful sanity check for CLOSED to be illegal, though.

@ableegoldman ableegoldman Jun 12, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I agree 100%, but at some point in the past we started checking for RESTORING and throwing IllegalStateException all over StandbyTask. I wanted to keep the changes here to a minimum and figured we should at least be consistent with the current pattern elsewhere

case CLOSED:
throw new IllegalStateException("Illegal state " + state() + " while suspending standby task " + id);

default:
throw new IllegalStateException("Unknown state " + state() + " while suspending standby task " + id);
}
}

Expand Down Expand Up @@ -172,10 +186,7 @@ public void closeDirty() {

@Override
public void closeAndRecycleState() {
suspend();
prepareCommit();

if (state() == State.CREATED || state() == State.SUSPENDED) {
if (state() == State.SUSPENDED) {
stateMgr.recycle();
} else {
throw new IllegalStateException("Illegal state " + state() + " while closing standby task " + id);
Expand All @@ -189,7 +200,6 @@ public void closeAndRecycleState() {

private void close(final boolean clean) {
switch (state()) {
case CREATED:
case SUSPENDED:
executeAndMaybeSwallow(
clean,
Expand All @@ -212,6 +222,7 @@ private void close(final boolean clean) {
log.trace("Skip closing since state is {}", state());
return;

case CREATED:
case RESTORING: // a StandbyTask is never in RESTORING state
case RUNNING:
throw new IllegalStateException("Illegal state " + state() + " while closing standby task " + id);
Expand Down
97 changes: 34 additions & 63 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,8 +107,6 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
private boolean commitNeeded = false;
private boolean commitRequested = false;

private Map<TopicPartition, Long> checkpoint = null;

public StreamTask(final TaskId id,
final Set<TopicPartition> partitions,
final ProcessorTopology topology,
Expand Down Expand Up @@ -250,14 +248,9 @@ public void completeRestoration() {
public void suspend() {
switch (state()) {
case CREATED:
case SUSPENDED:
Comment thread
ableegoldman marked this conversation as resolved.
Outdated
log.info("Skip suspending since state is {}", state());

break;

case RESTORING:
log.info("Suspended {}", state());
transitionTo(State.SUSPENDED);
log.info("Suspended restoring");

break;

Expand All @@ -272,6 +265,12 @@ public void suspend() {

break;

case SUSPENDED:
log.info("Skip suspending since state is {}", state());

break;


case CLOSED:
throw new IllegalStateException("Illegal state " + state() + " while suspending active task " + id);

Expand Down Expand Up @@ -342,7 +341,6 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit() {
case RUNNING:
case RESTORING:
case SUSPENDED:
maybeScheduleCheckpoint();
stateMgr.flush();
recordCollector.flush();

Expand Down Expand Up @@ -409,30 +407,38 @@ private Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
return committableOffsets;
}

/**
* This should only be called if the attempted commit succeeded for this task
*/
@Override
public void postCommit() {
commitRequested = false;
commitNeeded = false;

switch (state()) {
case RESTORING:
writeCheckpointIfNeed();
writeCheckpoint();

break;

case RUNNING:
if (!eosEnabled) { // if RUNNING, checkpoint only for non-eos
writeCheckpointIfNeed();
if (!eosEnabled) {
writeCheckpoint();
}

break;

case SUSPENDED:
writeCheckpointIfNeed();
// we cannot `clear()` the `PartitionGroup` in `suspend()` already, but only after committing,
// because otherwise we loose the partition-time information
/*
* We must clear the `PartitionGroup` only after committing, and not in `suspend()`,
* because otherwise we lose the partition-time information.
* We also must clear it when the task is revoked, and not in `close()`, as the consumer will clear
* its internal buffer when the corresponding partition is revoked but the task may be reassigned
*/
partitionGroup.clear();

@mjsax mjsax Jun 16, 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 think we still need to make this call -- in eager rebalancing, we suspend a task when we get a partition revoked. For this case, we "forget" the current offset within the consumer and thus need to clear the partition grouper. Otherwise, we might read the data a second time, if the partition is reassigned (what would violate EOS).

@ableegoldman ableegoldman Jun 16, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I see. So we should only clear it here, and not in close

Just curious, why do we "forget" the current offset? I mean, haven't we just committed the current offset before suspending (and if that failed we would close all tasks right away). Maybe I'm misunderstanding what you mean

@mjsax mjsax Jun 16, 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.

The consumer tracks offset internal, however, we buffer data in our internal queue. Thus, the offset tracked by the consumer, might be larger than the offset we commit (we take the offset we commit not from the consumer, but it's based on the records we did take out of the queue and processed).

In eager rebalancing, the consumer clears its internal state if a partition in revoked (and we only suspend the task), including the tracked offsets. If the partition in re-assigned, the consumer fetches the last committed offset to start fetching. Thus, if we don't clear the queue, we might fetch same data that is already in the queue a second time.

Does this make sense?

@guozhangwang guozhangwang Jun 16, 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.

Ah that's a good catch. Makes sense to me.

@ableegoldman ableegoldman Jun 16, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Got it, thanks for the explanation. I'll move it back to postCommit with a note

@ableegoldman ableegoldman Jun 16, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I'm not 100% certain that the Consumer does clear its internal buffer on revocation. At least, I couldn't find it in the code, but maybe I'm looking in the wrong place.

Not arguing we shouldn't clear the partition group here, was just wondering about this for my own sake. Hm

@mjsax mjsax Jun 16, 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.

Not 100% familiar with the consumer code, but in SubscriptionState#assignFromSubscribed new TopicPartitionState are created with position = null.

@ableegoldman ableegoldman Jun 16, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Ah that seems right. Thanks!


writeCheckpoint();

break;

case CREATED:
Expand Down Expand Up @@ -474,84 +480,47 @@ public void update(final Set<TopicPartition> topicPartitions, final Map<String,

@Override
public void closeAndRecycleState() {
suspend();
prepareCommit();
writeCheckpointIfNeed();

switch (state()) {
case CREATED:
case SUSPENDED:
stateMgr.recycle();
recordCollector.close();

break;

case RESTORING: // we should have transitioned to `SUSPENDED` already
case RUNNING: // we should have transitioned to `SUSPENDED` already
case CREATED:
case RESTORING:
case RUNNING:
case CLOSED:
throw new IllegalStateException("Illegal state " + state() + " while recycling active task " + id);
default:
throw new IllegalStateException("Unknown state " + state() + " while recycling active task " + id);
}

partitionGroup.clear();
closeTaskSensor.record();

transitionTo(State.CLOSED);

log.info("Closed clean and recycled state");
}

private void maybeScheduleCheckpoint() {
switch (state()) {
case RESTORING:
case SUSPENDED:
this.checkpoint = checkpointableOffsets();

break;

case RUNNING:
if (!eosEnabled) {
this.checkpoint = checkpointableOffsets();
}

break;

case CREATED:
case CLOSED:
throw new IllegalStateException("Illegal state " + state() + " while scheduling checkpoint for active task " + id);

default:
throw new IllegalStateException("Unknown state " + state() + " while scheduling checkpoint for active task " + id);
}
}

private void writeCheckpointIfNeed() {
private void writeCheckpoint() {
if (commitNeeded) {
log.error("Tried to write a checkpoint with pending uncommitted data, should complete the commit first.");
throw new IllegalStateException("A checkpoint should only be written if no commit is needed.");
}
if (checkpoint != null) {
stateMgr.checkpoint(checkpoint);
checkpoint = null;
}
stateMgr.checkpoint(checkpointableOffsets());
}

/**
* <pre>
* the following order must be followed:
* 1. checkpoint the state manager -- even if we crash before this step, EOS is still guaranteed
* 2. then if we are closing on EOS and dirty, wipe out the state store directory
* 3. finally release the state manager lock
* </pre>
* You must commit a task and checkpoint the state manager before closing as this will release the state dir lock
*/
private void close(final boolean clean) {
if (clean) {
executeAndMaybeSwallow(true, this::writeCheckpointIfNeed, "state manager checkpoint", log);
if (clean && commitNeeded) {
log.debug("Tried to close clean but there was pending uncommitted data, this means we failed to"
+ " commit and should close as dirty instead");
throw new StreamsException("Tried to close dirty task as clean");
}

switch (state()) {
case CREATED:
case RESTORING:
case SUSPENDED:
// first close state manager (which is idempotent) then close the record collector
// if the latter throws and we re-close dirty which would close the state manager again.
Expand All @@ -577,6 +546,8 @@ private void close(final boolean clean) {
log.trace("Skip closing since state is {}", state());
return;

case CREATED:
case RESTORING:
case RUNNING:
throw new IllegalStateException("Illegal state " + state() + " while closing active task " + id);

Expand Down
22 changes: 11 additions & 11 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,21 +56,21 @@ public interface Task {
* | | | |
* | v | |
* | +------+--------+ | |
* | | Suspended (3) | <---+ | //TODO Suspended(3) could be removed after we've stable on KIP-429
* | +------+--------+ |
* | | |
* | v |
* | +-----+-------+ |
* +----> | Closed (4) | -----------+
* +---> | Suspended (3) | ----+ | //TODO Suspended(3) could be removed after we've stable on KIP-429
* +------+--------+ |
* | |
* v |
* +-----+-------+ |
* | Closed (4) | -----------+
* +-------------+
* </pre>
*/
enum State {
CREATED(1, 4), // 0
RESTORING(2, 3, 4), // 1
RUNNING(3), // 2
SUSPENDED(1, 4), // 3
CLOSED(0); // 4, we allow CLOSED to transit to CREATED to handle corrupted tasks
CREATED(1, 3), // 0
RESTORING(2, 3), // 1
RUNNING(3), // 2
SUSPENDED(1, 4), // 3
CLOSED(0); // 4, we allow CLOSED to transit to CREATED to handle corrupted tasks
Comment thread
ableegoldman marked this conversation as resolved.
Outdated

private final Set<Integer> validTransitions = new HashSet<>();

Expand Down
Loading