fishfishfishfishaa commented on code in PR #8372:
URL: https://github.com/apache/paimon/pull/8372#discussion_r3529761460
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java:
##########
@@ -132,4 +136,17 @@ public DataStream<Committable> doWrite(
return written;
}
+
+ @Override
+ public DataStreamSink<?> doCommit(DataStream<Committable> written, String
commitUser) {
+ Options options = Options.fromMap(table.options());
+ if (options.get(SINK_COMMITTER_COORDINATOR_OPERATOR_ENABLED)
+ && !options.get(PRECOMMIT_COMPACT)) {
+ return written.sinkTo(new DiscardingSink<>())
Review Comment:
The coordinator path bypasses FlinkSink#doCommit, so the committer-operator
wrappers for auto-tag-for-savepoint and batch tag creation are not installed. I
added validation for sink.auto-tag-for-savepoint and batch runtime with
tag.creation-mode=BATCH when sink.committer-coordinator-operator.enabled is
true.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/PendingCheckpoint.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.coordinator;
+
+import org.apache.paimon.flink.sink.Committable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+/** Pending data and state for one checkpoint in PWC. */
+class PendingCheckpoint {
+
+ private final long checkpointId;
+ private final Map<Integer, SubtaskFileInfo> fileInfos;
+ private boolean staged;
+
+ PendingCheckpoint(long checkpointId) {
+ this.checkpointId = checkpointId;
+ this.fileInfos = new HashMap<>();
+ }
+
+ long checkpointId() {
+ return checkpointId;
+ }
+
+ boolean receive(int subtask, FileInfoRequest request, List<Committable>
committables) {
+ SubtaskFileInfo previous = fileInfos.get(subtask);
+ if (previous != null) {
+ if (previous.request().samePayload(request)) {
+ return false;
+ }
+ throw new IllegalStateException(
+ String.format(
+ "Different FileInfoRequest received for checkpoint
%d subtask %d.",
+ checkpointId, subtask));
+ }
+
+ fileInfos.put(subtask, new SubtaskFileInfo(request, committables));
+ return true;
+ }
+
+ void removeSubtask(int subtask) {
+ fileInfos.remove(subtask);
+ }
+
+ boolean isEmpty() {
+ return fileInfos.isEmpty();
+ }
+
+ boolean staged() {
+ return staged;
+ }
+
+ void markStaged() {
+ staged = true;
+ }
+
+ List<SubtaskFileInfo> fileInfos() {
+ return new ArrayList<>(fileInfos.values());
+ }
+
+ List<Committable> allCommittables() {
+ List<Committable> result = new ArrayList<>();
+ for (Integer subtask : new TreeSet<>(fileInfos.keySet())) {
+ result.addAll(fileInfos.get(subtask).committables());
+ }
+ return result;
+ }
+
+ List<Committable> committablesAfter(long checkpointId) {
+ List<Committable> result = new ArrayList<>();
+ for (Committable committable : allCommittables()) {
+ if (committable.checkpointId() > checkpointId) {
+ result.add(committable);
+ }
+ }
+ return result;
+ }
+
+ long maxWatermark() {
+ long watermark = Long.MIN_VALUE;
+ for (SubtaskFileInfo fileInfo : fileInfos.values()) {
+ watermark = Math.max(watermark, fileInfo.request().watermark());
Review Comment:
Thanks, agreed. I updated this to use the minimum writer watermark for the
checkpoint envelope, matching the previous single committer operator semantics.
The [1000, 0] case now commits watermark 0, and I added a regression test for
it.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/PaimonWriterCoordinator.java:
##########
@@ -0,0 +1,425 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink.coordinator;
+
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.Committer;
+import org.apache.paimon.flink.sink.TableWriteOperator;
+import org.apache.paimon.utils.ExceptionUtils;
+
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.operators.coordination.CoordinationRequest;
+import
org.apache.flink.runtime.operators.coordination.CoordinationRequestHandler;
+import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
+import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+import org.apache.flink.util.ThrowableCatchingRunnable;
+import org.apache.flink.util.function.ThrowingRunnable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * {@link OperatorCoordinator} for {@link TableWriteOperator}. It receives
writer file information
+ * and performs global commits in JobManager.
+ */
+public class PaimonWriterCoordinator implements OperatorCoordinator,
CoordinationRequestHandler {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PaimonWriterCoordinator.class);
+
+ private final PendingSubtask pendingSubtask;
+ private final CoordinatorExecutorThreadFactory coordinatorThreadFactory;
+ private final CompletableFuture<Void> finalCheckpointCompleted = new
CompletableFuture<>();
+
+ private final OperatorCoordinator.Context context;
+ private final CommitterCoordinator<Committable, ?> coordinator;
+ private final String initialCommitUser;
+
+ private @Nullable String commitUser;
+ private ScheduledExecutorService coordinatorExecutor;
+ private boolean started;
+ private boolean freshInstance = true;
+
+ public PaimonWriterCoordinator(
+ boolean streamingCheckpointEnabled,
+ String initialCommitUser,
+ Committer.Factory<Committable, ?> committerFactory,
+ OperatorCoordinator.Context context,
+ CoordinatorExecutorThreadFactory coordinatorThreadFactory,
+ Long endInputWatermark) {
+ this.context = context;
+ this.coordinatorThreadFactory = coordinatorThreadFactory;
+ this.initialCommitUser = initialCommitUser;
+ this.coordinator =
+ new CommitterCoordinator<>(
+ streamingCheckpointEnabled, committerFactory,
endInputWatermark);
+ this.pendingSubtask = new PendingSubtask(this.coordinator);
+ }
+
+ @Override
+ public void start() throws Exception {
+ OperatorID operatorId = context.getOperatorId();
+ LOG.info("Paimon writer coordinator starting, operatorId={}",
operatorId);
+ if (commitUser == null) {
+ commitUser = initialCommitUser;
+ }
+ started = true;
+ coordinatorExecutor = Executors.newScheduledThreadPool(1,
coordinatorThreadFactory);
+ int parallelism = context.currentParallelism();
+ coordinator.init(parallelism, commitUser);
+ pendingSubtask.init(parallelism);
+ }
+
+ @Override
+ public void executionAttemptReady(int subtask, int attemptNumber,
SubtaskGateway gateway) {
+ runInEventLoop(
+ () -> pendingSubtask.registerSubtask(subtask, attemptNumber,
gateway),
+ "registering subtask %d attempt %d",
+ subtask,
+ attemptNumber);
+ }
+
+ @Override
+ public void executionAttemptFailed(int subtask, int attemptNumber,
Throwable throwable) {
+ runInEventLoop(
+ () -> pendingSubtask.unregisterSubtask(subtask, attemptNumber,
throwable),
+ "unregistering subtask %d attempt %d",
+ subtask,
+ attemptNumber);
+ }
+
+ @Override
+ public void handleEventFromOperator(int subtask, int attemptNumber,
OperatorEvent event) {
+ freshInstance = false;
+ throw new UnsupportedOperationException(
+ "PWC only accepts file info through coordination requests.");
+ }
+
+ @Override
+ public CompletableFuture<CoordinationResponse> handleCoordinationRequest(
+ CoordinationRequest request) {
+ freshInstance = false;
+ if (request instanceof FileInfoRequest) {
+ return handleFileInfoRequest((FileInfoRequest) request);
+ }
+ CompletableFuture<CoordinationResponse> result = new
CompletableFuture<>();
+ result.completeExceptionally(
+ new IllegalArgumentException("Unsupported request type: " +
request.getClass()));
+ return result;
+ }
+
+ @Override
+ public void checkpointCoordinator(long checkpointId,
CompletableFuture<byte[]> result) {
+ freshInstance = false;
+ LOG.info("PWC snapshot commitUser={}, checkpointId={}", commitUser,
checkpointId);
+ checkState(commitUser != null, "PWC has not been started.");
+ result.complete(serializeCoordinatorState(commitUser));
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ freshInstance = false;
+ runInEventLoop(
+ () -> {
+
handleCommitResult(pendingSubtask.notifyCheckpointComplete(checkpointId));
+ if (coordinator.isEndInput()) {
+ finalCheckpointCompleted.complete(null);
+ }
+ },
+ "notifying checkpoint %d complete",
+ checkpointId);
+ if (coordinator.isEndInput()) {
+ try {
+ finalCheckpointCompleted.get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) {
+ runInEventLoop(
+ () -> pendingSubtask.notifyCheckpointAborted(checkpointId),
+ "notifying checkpoint %d aborted",
+ checkpointId);
+ }
+
+ @Override
+ public void resetToCheckpoint(long checkpointId, byte[] bytes) throws
Exception {
+ LOG.info("PWC resetToCheckpoint: checkpointId={}, fresh={}",
checkpointId, freshInstance);
+ if (freshInstance && checkpointId >= 0) {
+ checkState(!started, "PWC can only be restored before it is
started.");
Review Comment:
Good catch. I changed the writer coordinator provider to extend
`RecreateOnResetOperatorCoordinator.Provider`, same as
`DataStatisticsCoordinatorProvider`, so Flink recreates the coordinator on
checkpoint reset and restores state before `start()`. This should avoid reusing
an already-started coordinator during JM/global failover.
--
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]
