becketqin commented on a change in pull request #13784:
URL: https://github.com/apache/flink/pull/13784#discussion_r514953305
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
##########
@@ -20,6 +20,7 @@
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.state.CheckpointListener;
Review comment:
Generally speaking, the deprecation of an API should start with all the
internal invocations migration. So having the internal calls migrated in the
same patch seems a natural part of the deprecation.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/OperatorCoordinatorCheckpointContext.java
##########
@@ -27,15 +28,33 @@
* This context is the interface through which the {@link
CheckpointCoordinator} interacts with an
* {@link OperatorCoordinator} during checkpointing and checkpoint restoring.
*/
-public interface OperatorCoordinatorCheckpointContext extends OperatorInfo {
+public interface OperatorCoordinatorCheckpointContext extends OperatorInfo,
CheckpointListener {
void checkpointCoordinator(long checkpointId, CompletableFuture<byte[]>
result) throws Exception;
void afterSourceBarrierInjection(long checkpointId);
void abortCurrentTriggering();
- void checkpointComplete(long checkpointId);
+ /**
+ * This method does not throw exception because the operator
coordinator implementation
+ * is supposed to call {@link
OperatorCoordinator.Context#failJob(Throwable)} in case
+ * of receiving an exception.
+ *
+ * @param checkpointId The ID of the checkpoint that has been aborted.
+ */
+ @Override
+ void notifyCheckpointComplete(long checkpointId);
+
+ /**
+ * This method does not throw exception because the operator
coordinator implementation
+ * is supposed to call {@link
OperatorCoordinator.Context#failJob(Throwable)} in case
+ * of receiving an exception.
+ *
+ * @param checkpointId The ID of the checkpoint that has been aborted.
+ */
+ @Override
+ default void notifyCheckpointAborted(long checkpointId) {}
Review comment:
It is because most implementations don't need to worry about the
checkpoint abortion.
##########
File path:
flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReader.java
##########
@@ -58,7 +60,7 @@
*
* @return the state of the source.
*/
- List<SplitT> snapshotState();
+ List<SplitT> snapshotState(long checkpointId);
Review comment:
Generally speaking, I am a fan of keeping backwards compatibility.
However, in this case, it seems biting the bullet once is better.
The suggested solution would be able to keep the backwards compatibility for
now. However, it would also force the new implementations to implement
`snapshotState()` as well, which is a little ugly and they would have to change
the code again after we deprecate this old method.
One way to improve that is to let both `snapshotState()` and
`snapshotState(checkpointId)` have a default implementation. So the new
implementations can just implement `snapshotCheckpoint(checkpointId)`. But that
is a little misleading because we actually want people to think about what to
do during checkpoint.
Given that:
1. the API is marked as public evolving and there should not be many users
implementing this API anyways, and
2. We have other changes breaking backwards compatibility in the Source in
1.12.
Personally I think it is better to just change it to the final state in one
shot.
----------------------------------------------------------------
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:
[email protected]
