1996fanrui commented on code in PR #24653:
URL: https://github.com/apache/flink/pull/24653#discussion_r1582655229
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java:
##########
@@ -234,16 +230,15 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
private static ChannelStateWriter openChannelStateWriter(
String taskName,
- CheckpointStorage checkpointStorage,
- Environment env,
- int maxSubtasksPerChannelStateFile) {
+ CheckpointStorageWorkerView checkpointStorageWorkerView,
+ Environment env) {
return new ChannelStateWriterImpl(
env.getJobVertexId(),
taskName,
env.getTaskInfo().getIndexOfThisSubtask(),
- checkpointStorage,
+ checkpointStorageWorkerView,
env.getChannelStateExecutorFactory(),
- maxSubtasksPerChannelStateFile);
+ 1 /* one logical channel state file/executor per subtask */);
Review Comment:
If we hard code `maxSubtasksPerChannelStateFile` is 1, it means
`execution.checkpointing.unaligned.max-subtasks-per-channel-state-file` doesn't
take effect after this PR, right?
I'm thinking could we keep the
`execution.checkpointing.unaligned.max-subtasks-per-channel-state-file` first?
and deprecate it after `unified file merging mechanism` is stable.
WDYT?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequestDispatcherImpl.java:
##########
@@ -213,14 +205,13 @@ private void failAndClearWriter(
writer = null;
}
- private ChannelStateCheckpointWriter buildWriter(CheckpointStartRequest
request)
+ protected ChannelStateCheckpointWriter buildWriter(CheckpointStartRequest
request)
Review Comment:
Is private enough?
--
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
