ableegoldman commented on a change in pull request #8248:
URL: https://github.com/apache/kafka/pull/8248#discussion_r418858652
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
##########
@@ -505,4 +550,35 @@ private StateStoreMetadata findStore(final TopicPartition
changelogPartition) {
return found.isEmpty() ? null : found.get(0);
}
+
+ void prepareForRecycle() {
+ log.debug("Preparing to recycle state for {} task {}.", taskType,
taskId);
+
+ if (recyclingState) {
+ throw new IllegalStateException("Attempted to re-recycle state
without completing first recycle");
+ }
+ recyclingState = true;
+ }
+
+ void recycleState() {
+ log.debug("Completed recycling state for formerly {} task {}.",
taskType, taskId);
+
+ if (!recyclingState) {
+ throw new IllegalStateException("Attempted to complete recycle but
state is not currently being recycled");
+ }
+ recyclingState = false;
+
+ if (taskType == TaskType.ACTIVE) {
+ taskType = TaskType.STANDBY;
+ } else {
+ taskType = TaskType.ACTIVE;
+ }
Review comment:
Alright I reworked things slightly here. With nulling out the task type
during close/recycle we can technically get away with not even having the
`recyclingState` flag at all. But I thought it might be a good idea to keep the
two distinct to avoid future messes
----------------------------------------------------------------
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:
us...@infra.apache.org
