You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/06/09 16:34:14 UTC
[GitHub] [kafka] abbccdda commented on a change in pull request #8833: KAFKA-9441: remove prepareClose() to simplify task management
abbccdda commented on a change in pull request #8833:
URL: https://github.com/apache/kafka/pull/8833#discussion_r436987081
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -534,62 +482,38 @@ public void closeAndRecycleState() {
log.info("Closed clean and recycled state");
}
- /**
- * <pre>
- * the following order must be followed:
- * 1. first close topology to make sure all cached records in the topology are processed
- * 2. then flush the state, send any left changelog records
- * 3. then flush the record collector
- * </pre>
- *
- * @param clean shut down cleanly (ie, incl. flush) if {@code true} --
- * otherwise, just close open resources
- * @throws TaskMigratedException if the task producer got fenced (EOS)
- */
- private void prepareClose(final boolean clean) {
- // Reset any previously scheduled checkpoint.
- checkpoint = null;
-
+ private void maybeScheduleCheckpoint() {
switch (state()) {
- case CREATED:
- // the task is created and not initialized, just re-write the checkpoint file
- scheduleCheckpoint(emptyMap());
+ case RESTORING:
+ this.checkpoint = checkpointableOffsets();
+
break;
case RUNNING:
- closeTopology(clean);
-
- if (clean) {
- stateMgr.flush();
- recordCollector.flush();
- scheduleCheckpoint(checkpointableOffsets());
- } else {
- executeAndMaybeSwallow(false, stateMgr::flush, "state manager flush", log);
+ if (!eosEnabled) {
+ this.checkpoint = checkpointableOffsets();
}
break;
- case RESTORING:
- executeAndMaybeSwallow(clean, stateMgr::flush, "state manager flush", log);
- scheduleCheckpoint(emptyMap());
+ case SUSPENDED:
+ this.checkpoint = Collections.emptyMap();
break;
- case SUSPENDED:
+ case CREATED:
case CLOSED:
- // not need to checkpoint, since when suspending we've already committed the state
- break;
+ throw new IllegalStateException("Illegal state " + state() + " while closing active task " + id);
default:
- throw new IllegalStateException("Unknown state " + state() + " while prepare closing active task " + id);
+ throw new IllegalStateException("Unknown state " + state() + " while closing active task " + id);
Review comment:
similar here
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
##########
@@ -260,8 +220,8 @@ private void close(final boolean clean) {
log.trace("Skip closing since state is {}", state());
return;
- case RESTORING:
- case SUSPENDED:
+ case RESTORING: // a StandbyTask in never in RESTORING state
Review comment:
Seems like we could use a separate illegal state exception here for `RESTORING` as we should never hit it.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -534,62 +482,38 @@ public void closeAndRecycleState() {
log.info("Closed clean and recycled state");
}
- /**
- * <pre>
- * the following order must be followed:
- * 1. first close topology to make sure all cached records in the topology are processed
- * 2. then flush the state, send any left changelog records
- * 3. then flush the record collector
- * </pre>
- *
- * @param clean shut down cleanly (ie, incl. flush) if {@code true} --
- * otherwise, just close open resources
- * @throws TaskMigratedException if the task producer got fenced (EOS)
- */
- private void prepareClose(final boolean clean) {
- // Reset any previously scheduled checkpoint.
- checkpoint = null;
-
+ private void maybeScheduleCheckpoint() {
switch (state()) {
- case CREATED:
- // the task is created and not initialized, just re-write the checkpoint file
- scheduleCheckpoint(emptyMap());
+ case RESTORING:
+ this.checkpoint = checkpointableOffsets();
+
break;
case RUNNING:
- closeTopology(clean);
-
- if (clean) {
- stateMgr.flush();
- recordCollector.flush();
- scheduleCheckpoint(checkpointableOffsets());
- } else {
- executeAndMaybeSwallow(false, stateMgr::flush, "state manager flush", log);
+ if (!eosEnabled) {
+ this.checkpoint = checkpointableOffsets();
}
break;
- case RESTORING:
- executeAndMaybeSwallow(clean, stateMgr::flush, "state manager flush", log);
- scheduleCheckpoint(emptyMap());
+ case SUSPENDED:
+ this.checkpoint = Collections.emptyMap();
break;
- case SUSPENDED:
+ case CREATED:
case CLOSED:
- // not need to checkpoint, since when suspending we've already committed the state
- break;
+ throw new IllegalStateException("Illegal state " + state() + " while closing active task " + id);
Review comment:
Should update the exception text like `while scheduling checkpoint`
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -246,82 +245,39 @@ public void completeRestoration() {
}
}
- /**
- * <pre>
- * the following order must be followed:
- * 1. first close topology to make sure all cached records in the topology are processed
- * 2. then flush the state, send any left changelog records
- * 3. then flush the record collector
- * 4. then commit the record collector -- for EOS this is the synchronization barrier
- * 5. then checkpoint the state manager -- even if we crash before this step, EOS is still guaranteed
- * </pre>
- *
- * @throws TaskMigratedException if committing offsets failed (non-EOS)
- * or if the task producer got fenced (EOS)
- */
@Override
- public void prepareSuspend() {
- switch (state()) {
- case CREATED:
- case SUSPENDED:
- // do nothing
- log.trace("Skip prepare suspending since state is {}", state());
-
- break;
-
- case RESTORING:
- stateMgr.flush();
- log.info("Prepare suspending restoring");
-
- break;
-
- case RUNNING:
- closeTopology(true);
-
- stateMgr.flush();
- recordCollector.flush();
-
- log.info("Prepare suspending running");
-
- break;
-
- case CLOSED:
- throw new IllegalStateException("Illegal state " + state() + " while suspending active task " + id);
-
- default:
- throw new IllegalStateException("Unknown state " + state() + " while suspending active task " + id);
- }
+ public void suspendDirty() {
+ log.info("Suspending dirty");
+ suspend(false);
}
@Override
- public void suspend() {
+ public void suspendCleanAndPrepareCommit() {
Review comment:
Although this holds true for now, I'm thinking whether in long term a `commit` will always follow `suspendClean`, what's the downside of having these two logics separate?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -389,30 +346,27 @@ public void prepareCommit() {
@Override
public void postCommit() {
switch (state()) {
+ case RESTORING:
case RUNNING:
- commitNeeded = false;
+ case SUSPENDED:
commitRequested = false;
+ commitNeeded = false;
- if (!eosEnabled) {
- stateMgr.checkpoint(checkpointableOffsets());
+ if (state() == State.RESTORING) {
Review comment:
Could we merge the two if-else here?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
##########
@@ -108,13 +107,20 @@ public void completeRestoration() {
}
@Override
- public void prepareSuspend() {
- log.trace("No-op prepareSuspend with state {}", state());
+ public void suspendDirty() {
+ log.trace("No-op suspend dirty with state {}", state());
+ if (state() == State.RUNNING) {
+ transitionTo(State.SUSPENDED);
+ }
}
@Override
- public void suspend() {
- log.trace("No-op suspend with state {}", state());
+ public void suspendCleanAndPrepareCommit() {
+ log.trace("No-op suspend clean with state {}", state());
+ if (state() == State.RUNNING) {
Review comment:
Similar here, maybe we could leverage `transitionTo` to help throw the exception.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -506,19 +455,18 @@ public void update(final Set<TopicPartition> topicPartitions, final ProcessorTop
@Override
public void closeAndRecycleState() {
- prepareClose(true);
-
+ suspendCleanAndPrepareCommit();
writeCheckpointIfNeed();
switch (state()) {
case CREATED:
- case RUNNING:
- case RESTORING:
case SUSPENDED:
stateMgr.recycle();
recordCollector.close();
break;
+ case RESTORING: // we should have transitioned to `SUSPENDED` already
Review comment:
Prefer to throw different illegal state exception here than making comments
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -534,62 +482,38 @@ public void closeAndRecycleState() {
log.info("Closed clean and recycled state");
}
- /**
- * <pre>
- * the following order must be followed:
- * 1. first close topology to make sure all cached records in the topology are processed
- * 2. then flush the state, send any left changelog records
- * 3. then flush the record collector
- * </pre>
- *
- * @param clean shut down cleanly (ie, incl. flush) if {@code true} --
- * otherwise, just close open resources
- * @throws TaskMigratedException if the task producer got fenced (EOS)
- */
- private void prepareClose(final boolean clean) {
- // Reset any previously scheduled checkpoint.
- checkpoint = null;
-
+ private void maybeScheduleCheckpoint() {
switch (state()) {
- case CREATED:
- // the task is created and not initialized, just re-write the checkpoint file
- scheduleCheckpoint(emptyMap());
+ case RESTORING:
+ this.checkpoint = checkpointableOffsets();
+
break;
case RUNNING:
- closeTopology(clean);
-
- if (clean) {
- stateMgr.flush();
- recordCollector.flush();
- scheduleCheckpoint(checkpointableOffsets());
- } else {
- executeAndMaybeSwallow(false, stateMgr::flush, "state manager flush", log);
+ if (!eosEnabled) {
+ this.checkpoint = checkpointableOffsets();
}
break;
- case RESTORING:
- executeAndMaybeSwallow(clean, stateMgr::flush, "state manager flush", log);
- scheduleCheckpoint(emptyMap());
+ case SUSPENDED:
+ this.checkpoint = Collections.emptyMap();
break;
- case SUSPENDED:
+ case CREATED:
case CLOSED:
- // not need to checkpoint, since when suspending we've already committed the state
- break;
+ throw new IllegalStateException("Illegal state " + state() + " while closing active task " + id);
default:
- throw new IllegalStateException("Unknown state " + state() + " while prepare closing active task " + id);
+ throw new IllegalStateException("Unknown state " + state() + " while closing active task " + id);
}
}
- private void scheduleCheckpoint(final Map<TopicPartition, Long> checkpoint) {
- this.checkpoint = checkpoint;
- }
-
private void writeCheckpointIfNeed() {
+ if (commitNeeded) {
+ throw new IllegalStateException("A checkpoint should only be written if now commit is needed.");
Review comment:
`if commit is needed`
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -246,82 +245,39 @@ public void completeRestoration() {
}
}
- /**
- * <pre>
- * the following order must be followed:
- * 1. first close topology to make sure all cached records in the topology are processed
- * 2. then flush the state, send any left changelog records
- * 3. then flush the record collector
- * 4. then commit the record collector -- for EOS this is the synchronization barrier
- * 5. then checkpoint the state manager -- even if we crash before this step, EOS is still guaranteed
- * </pre>
- *
- * @throws TaskMigratedException if committing offsets failed (non-EOS)
- * or if the task producer got fenced (EOS)
- */
@Override
- public void prepareSuspend() {
- switch (state()) {
- case CREATED:
- case SUSPENDED:
- // do nothing
- log.trace("Skip prepare suspending since state is {}", state());
-
- break;
-
- case RESTORING:
- stateMgr.flush();
- log.info("Prepare suspending restoring");
-
- break;
-
- case RUNNING:
- closeTopology(true);
-
- stateMgr.flush();
- recordCollector.flush();
-
- log.info("Prepare suspending running");
-
- break;
-
- case CLOSED:
- throw new IllegalStateException("Illegal state " + state() + " while suspending active task " + id);
-
- default:
- throw new IllegalStateException("Unknown state " + state() + " while suspending active task " + id);
- }
+ public void suspendDirty() {
+ log.info("Suspending dirty");
+ suspend(false);
}
@Override
- public void suspend() {
+ public void suspendCleanAndPrepareCommit() {
+ log.info("Suspending clean");
+ suspend(true);
+ }
+
+ @SuppressWarnings("fallthrough")
+ private void suspend(final boolean clean) {
switch (state()) {
case CREATED:
case SUSPENDED:
// do nothing
Review comment:
This comment could be removed IMHO.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -389,30 +346,27 @@ public void prepareCommit() {
@Override
public void postCommit() {
switch (state()) {
+ case RESTORING:
case RUNNING:
- commitNeeded = false;
+ case SUSPENDED:
commitRequested = false;
+ commitNeeded = false;
- if (!eosEnabled) {
- stateMgr.checkpoint(checkpointableOffsets());
+ if (state() == State.RESTORING) {
+ writeCheckpointIfNeed();
+ } else if (!eosEnabled) { // if RUNNING, checkpoint only for non-eos
Review comment:
Seems applied to both RUNNING and SUSPENDED?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
##########
@@ -260,8 +220,8 @@ private void close(final boolean clean) {
log.trace("Skip closing since state is {}", state());
return;
- case RESTORING:
- case SUSPENDED:
+ case RESTORING: // a StandbyTask in never in RESTORING state
Review comment:
a standby task is never in
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
##########
@@ -108,13 +107,20 @@ public void completeRestoration() {
}
@Override
- public void prepareSuspend() {
- log.trace("No-op prepareSuspend with state {}", state());
+ public void suspendDirty() {
+ log.trace("No-op suspend dirty with state {}", state());
+ if (state() == State.RUNNING) {
Review comment:
Do we want to throw here if the current state is `CLOSED`?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -891,6 +821,9 @@ private void initializeTopology() {
}
private void closeTopology(final boolean clean) {
+ if (state() != State.RUNNING) {
Review comment:
I could see we are trying to maintain the same behavior, but still why a restoring task won't need to close topology?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -389,30 +346,27 @@ public void prepareCommit() {
@Override
public void postCommit() {
switch (state()) {
+ case RESTORING:
case RUNNING:
- commitNeeded = false;
+ case SUSPENDED:
commitRequested = false;
+ commitNeeded = false;
- if (!eosEnabled) {
- stateMgr.checkpoint(checkpointableOffsets());
+ if (state() == State.RESTORING) {
+ writeCheckpointIfNeed();
+ } else if (!eosEnabled) { // if RUNNING, checkpoint only for non-eos
+ writeCheckpointIfNeed();
}
- log.debug("Committed");
-
- break;
-
- case RESTORING:
- commitNeeded = false;
- commitRequested = false;
-
- stateMgr.checkpoint(checkpointableOffsets());
+ if (state() == State.SUSPENDED) {
+ partitionGroup.clear();
Review comment:
The `partitionGroup.clear` and `partitionGroup.close` are interchangeable right now, should we just consolidate both?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org