AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1756482710
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -160,7 +169,19 @@ public void initializeState(StateInitializationContext
context) throws Exception
legacyCommitterState.clear();
}
}
- sinkWriter = writerStateHandler.createWriter(initContext, context);
+
+ // use union state to ensure that rescaling works correctly
+ this.endOfInputState =
+
context.getOperatorStateStore().getUnionListState(END_OF_INPUT_STATE_DESC);
Review Comment:
We can and need to. Think about the committables inside the
CommitterOperator. We absolutely need to track them. EOI just means
`processElement` isn't called again and you can't call collector afterwards.
--
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.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
