rkhachatryan commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r676745053
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
##########
@@ -43,7 +43,7 @@
* @param <OUT> The output type of the operator
*/
@PublicEvolving
-public interface StreamOperator<OUT> extends CheckpointListener, KeyContext,
Serializable {
+public interface StreamOperator<OUT> extends InternalCheckpointListener,
KeyContext, Serializable {
Review comment:
It seems wrong to me that a `@PublicEvolving` interface extends
`@Internal` interface.
The root cause I think is in the `StreamOperator` hierarchy (but that's a
different story).
Semantically, I think operators should only be aware of checkpoint
completion (to emit data for example); their backends should in addition know
about the subsumption.
I see the following solutions to notify state backends about the subsumption
1. Modify `CheckpointListener` interface instead of introducing
`InternalCheckpointListener` as you proposed initially
2. In `StreamOperatorWrapper`, get `KeyedStateBackend` from the operator
(if it's `AbstractStreamOperator`) and notify it directly (if it's
`InternalCheckpointListener`):
```
// SubtaskCheckpointCoordinatorImpl.notifyCheckpointSubsumed:
// loop through the chain:
operatorWrapper.notifyCheckpointSubsumed(checkpointId);
// StreamOperatorWrapper.notifyCheckpointSubsumed:
// .. casts omitted ...
`getStreamOperator().getKeyedStateBackend().notifyCheckpointSubsumed(checkpointId);`
```
3. Same, but via `StreamOperatorStateHandler`
WDYT?
cc: @pnowojski
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
