ableegoldman commented on a change in pull request #8248:
URL: https://github.com/apache/kafka/pull/8248#discussion_r418850068
##########
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:
That makes sense to me, I'll try it out
----------------------------------------------------------------
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
