ifndef-SleePy commented on code in PR #8671:
URL: https://github.com/apache/paimon/pull/8671#discussion_r4271081240
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java:
##########
@@ -82,6 +88,16 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
@Nullable private final SavepointTagger.Factory savepointTaggerFactory;
private final WriterCommittables[] subtaskCommittables;
+ // Successful commit coverage, copied into each later coordinator
checkpoint.
+ private final long[] terminalCoveredBy;
+ // Permission to finish before commit, rebuilt from candidate reports
after global recovery.
+ // The last terminal candidate never receives this permission: it guards
finalization.
+ private final long[] earlyReleaseCheckpoint;
+ // Scheduler callbacks invalidate targets immediately, even while a commit
is in flight.
+ private final Map<Integer, SubtaskGateway> releaseTargets = new
ConcurrentHashMap<>();
+ // Required writers still missing their restore contribution;
executor-confined.
+ private final BitSet requiredRestoreWriters;
+ private boolean recoveryInitialized;
Review Comment:
* Are all these additional state fields necessary? If so, could we keep the
related state and transitions together, perhaps in a dedicated class or method?
Their relationships are currently spread across several methods, which makes
the code difficult to follow and maintain.
* Could we keep all access to `releaseTargets` on `commitExecutor` and avoid
using a `ConcurrentHashMap` here?
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java:
##########
@@ -94,26 +110,44 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
private long restoredCheckpointId = OperatorCoordinator.NO_CHECKPOINT;
private byte[] restoredCheckpointData;
+ // Accessed only on the commit executor. Latch before notifying Flink of
failure.
+ @Nullable private Throwable fatalFailure;
+
private State state;
private Committer<Committable, ManifestCommittable> committer;
private String commitUser;
private MemoryBackendStateStore stateStore;
// Built in initializeAfterRestore once commitUser is known; null when
auto-tag is disabled.
@Nullable private SavepointTagger savepointTagger;
+ private final Long endInputWatermark;
+ private ListState<Long> processedWatermarkState;
+ @Nullable private Long lastProcessedWatermark;
+ private boolean globalFinalizationCompleted;
+
+ @VisibleForTesting
+ static final String PROCESSED_WATERMARK_STATE =
"coordinator-last-processed-watermark";
Review Comment:
Please see my comment above. This PR introduces more than 10 new fields to
this class.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperator.java:
##########
@@ -96,6 +102,14 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
private transient CheckpointCommittablesSerializer stateSerializer;
private transient TypeSerializer<CheckpointCommittables> eventSerializer;
+ private boolean endOfInput;
+ private boolean terminalCaptured;
+ private final transient MailboxExecutor taskMailbox;
+ private final int subtask;
+ private final int attemptNumber;
+ private long requiredReleaseCheckpoint = -1;
+ private long releasedCheckpoint = -1;
+ private boolean waitingForRelease;
Review Comment:
Please see my comment in `CommittingWriteOperatorCoordinator`. Their
relationships are currently spread across several methods, it makes the code
difficult to maintain.
--
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]
