Skip to content

KAFKA-9441: remove prepareClose() to simplify task management #8833

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 8 commits into from
Jun 11, 2020
Merged

KAFKA-9441: remove prepareClose() to simplify task management #8833

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
139bceb
KAFKA-9441: remove closePrepare() to simplify task management
mjsax Jun 8, 2020
1be5d92
Add missing tests
mjsax Jun 8, 2020
54f24e4
Github comments
mjsax Jun 9, 2020
120fb8d
Rebased
mjsax Jun 9, 2020
141c4d0
Github comments
mjsax Jun 9, 2020
c4b2666
Github comments
mjsax Jun 10, 2020
9e98592
Github comments
mjsax Jun 10, 2020
329e187
Github comments
mjsax Jun 10, 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
4 changes: 0 additions & 4 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/PartitionGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,10 +243,6 @@ boolean allPartitionsBuffered() {
return allBuffered;
}

void close() {
clear();
}

void clear() {
for (final RecordQueue queue : partitionQueues.values()) {
queue.clear();
Expand Down
92 changes: 22 additions & 70 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 @@ -17,13 +17,13 @@
package org.apache.kafka.streams.processor.internals;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskMigratedException;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.processor.internals.metrics.ThreadMetrics;
Expand Down Expand Up @@ -99,6 +99,8 @@ public void initializeIfNeeded() {
processorContext.initialize();

log.info("Initialized");
} else if (state() == State.RESTORING) {
throw new IllegalStateException("Illegal state " + state() + " while initializing standby task " + id);
}
}

Expand All @@ -107,18 +109,21 @@ public void completeRestoration() {
throw new IllegalStateException("Standby task " + id + " should never be completing restoration");
}

@Override
public void prepareSuspend() {
log.trace("No-op prepareSuspend with state {}", state());
}

@Override
public void suspend() {
log.trace("No-op suspend with state {}", state());
if (state() == State.RUNNING) {

@abbccdda abbccdda Jun 8, 2020

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Do we want to throw here if the current state is CLOSED?

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

My proposal is, to keep the methods idempotent.

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

nit: we can throw illegal-state if the state() == RESTORING since it should never happen.

@ableegoldman ableegoldman Jun 10, 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.

Why not just make suspend a no-op if the task is RESTORING? That seems more in line with how we handle things elsewhere

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

For StandbyTasks, we never restore. When we do the state transition, we away make two transitions directly after each other from CREATE -> RESTORING -> RUNNING -- thus, state RESTORING is an invalid state for standby tasks.

@ableegoldman ableegoldman Jun 10, 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.

Right, by "check for RESTORING" I meant "throw an exception if state is restoring". It seems odd to check for RESTORING during suspend but not in any other StandbyTask method. Either it can never be in RESTORING and we are completely sure of that, and shouldn't check for RESTORING, or we should always check whether it's RESTORING and not just during suspend (eg also in postCommit)

@ableegoldman ableegoldman Jun 10, 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.

Just to clarify, I would support doing the former, ie don't check whether it's RESTORING here at all. But we should at least be consistent

@mjsax mjsax Jun 10, 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, I see. I guess we check it in almost all method though. (we just missed initializeIfNeeded and resume() -- will add it there).

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

We do this check implicitly for some case already, ie:

if (RUNNING) {
} else {
  throw
}

ie, only RUNNING is a valid state, and all others are invalid. Thus, it seems to be consistent if we add those checks elsewhere (or, what would be odd, exclude RESTORING from those implicit checks).

@ableegoldman ableegoldman Jun 10, 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.

Cool 👍

transitionTo(State.SUSPENDED);
} else if (state() == State.RESTORING) {
throw new IllegalStateException("Illegal state " + state() + " while suspending standby task " + id);
}
}

@Override
public void resume() {
if (state() == State.RESTORING) {
throw new IllegalStateException("Illegal state " + state() + " while resuming standby task " + id);
}
log.trace("No-op resume with state {}", state());
}

Expand All @@ -129,18 +134,20 @@ public void resume() {
* @throws StreamsException fatal error, should close the thread
*/
@Override
public void prepareCommit() {
if (state() == State.RUNNING) {
public Map<TopicPartition, OffsetAndMetadata> prepareCommit() {
if (state() == State.RUNNING || state() == State.SUSPENDED) {
stateMgr.flush();
log.info("Task ready for committing");
} else {
throw new IllegalStateException("Illegal state " + state() + " while preparing standby task " + id + " for committing ");
}

return Collections.emptyMap();
}

@Override
public void postCommit() {
if (state() == State.RUNNING) {
if (state() == State.RUNNING || state() == State.SUSPENDED) {
// since there's no written offsets we can checkpoint with empty map,
// and the state current offset would be used to checkpoint
stateMgr.checkpoint(Collections.emptyMap());
Expand All @@ -151,73 +158,24 @@ public void postCommit() {
}
}

@Override
public void prepareCloseClean() {
prepareClose(true);

log.info("Prepared clean close");
}

@Override
public void prepareCloseDirty() {
prepareClose(false);

log.info("Prepared dirty close");
}

/**
* 1. commit if we are running and clean close;
* 2. close the state manager.
*
* @throws TaskMigratedException all the task has been migrated
* @throws StreamsException fatal error, should close the thread
*/
private void prepareClose(final boolean clean) {

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

This logic is now followed in suspendCleanAndPrepareCommit() that must be called before a task can be closed.

switch (state()) {
case CREATED:
case CLOSED:
log.trace("Skip prepare closing since state is {}", state());
return;

case RUNNING:
if (clean) {
stateMgr.flush();
}

break;

case RESTORING:
case SUSPENDED:
throw new IllegalStateException("Illegal state " + state() + " while closing standby task " + id);

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

@Override
public void closeClean() {
close(true);

log.info("Closed clean");
}

@Override
public void closeDirty() {
close(false);

log.info("Closed dirty");
}

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

@ableegoldman ableegoldman Jun 10, 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.

Why call prepareCommit (or suspend for that matter)?

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

Both now do what prepareClose() did before.

@ableegoldman ableegoldman Jun 11, 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 just mean, why not inline that? I'm just imagining coming back to this code in a few months and wondering why we need to suspend a task before recycling, or why we call prepareCommit but don't then actually commit, etc

@ableegoldman ableegoldman Jun 11, 2020 •
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.

Nevermind, I see that's the pattern we follow everywhere else


if (state() == State.CREATED || state() == State.RUNNING) {
// since there's no written offsets we can checkpoint with empty map,
// and the state current offset would be used to checkpoint
stateMgr.checkpoint(Collections.emptyMap());
offsetSnapshotSinceLastCommit = new HashMap<>(stateMgr.changelogOffsets());
if (state() == State.CREATED || state() == State.SUSPENDED) {

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

The comment below is not accurate anymore: we do not write checkpoint during recycle actually.

EDIT: actually, the updated offsetSnapshotSinceLastCommit seems not used since after this function we would create a new StreamTask and in between we do not check if commitNeeded at all. Could we remove line 175 then?

stateMgr.recycle();
} else {
throw new IllegalStateException("Illegal state " + state() + " while closing standby task " + id);
Expand All @@ -232,13 +190,7 @@ public void closeAndRecycleState() {
private void close(final boolean clean) {
switch (state()) {
case CREATED:
case RUNNING:
if (clean) {
// since there's no written offsets we can checkpoint with empty map,
// and the state current offset would be used to checkpoint
stateMgr.checkpoint(Collections.emptyMap());
offsetSnapshotSinceLastCommit = new HashMap<>(stateMgr.changelogOffsets());
}
case SUSPENDED:

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

Ditto for line 195: we do not need to update the snapshot since we are closing the task already.

executeAndMaybeSwallow(
clean,
() -> StateManagerUtil.closeStateManager(
Expand All @@ -260,8 +212,8 @@ private void close(final boolean clean) {
log.trace("Skip closing since state is {}", state());
return;

case RESTORING:
case SUSPENDED:
case RESTORING: // a StandbyTask is never in RESTORING state
case RUNNING:
throw new IllegalStateException("Illegal state " + state() + " while closing standby task " + id);

default:
Expand Down
Loading