ableegoldman commented on a change in pull request #8248:
URL: https://github.com/apache/kafka/pull/8248#discussion_r418885536
##########
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:
Actually turns out nulling it during close causes some tests to fail 😕
Probably the changelog reader. I'll just revert that part for now and look
into it next week
----------------------------------------------------------------
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
