diff --git a/docs/generated/flink_connector_configuration.html b/docs/generated/flink_connector_configuration.html index 0f2786b038dc..a80c819be02e 100644 --- a/docs/generated/flink_connector_configuration.html +++ b/docs/generated/flink_connector_configuration.html @@ -266,6 +266,12 @@ Boolean Allow sink committer and writer operator to be chained together + +
sink.coordinator-commit.enabled
+ false + Boolean + If true, run the Paimon committer inside the Flink JobManager via an OperatorCoordinator. This decouples commit from any single TaskManager subtask so that region failover does not have to restart the whole pipeline. Only supports unaware-bucket append tables in streaming mode with checkpointing enabled; unsupported configurations fail during sink planning. +
sink.cross-partition.managed-memory
256 mb diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java index 678471ea337c..7799acc2d2bd 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java @@ -486,6 +486,17 @@ public class FlinkConnectorOptions { "Commit listener will be called after a successful commit. This option list custom commit " + "listener identifiers separated by comma."); + public static final ConfigOption SINK_COORDINATOR_COMMIT_ENABLED = + key("sink.coordinator-commit.enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "If true, run the Paimon committer inside the Flink JobManager via an " + + "OperatorCoordinator. This decouples commit from any single TaskManager " + + "subtask so that region failover does not have to restart the whole pipeline. " + + "Only supports unaware-bucket append tables in streaming mode with " + + "checkpointing enabled; unsupported configurations fail during sink planning."); + public static final ConfigOption SINK_WRITER_COORDINATOR_ENABLED = key("sink.writer-coordinator.enabled") .booleanType() diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperator.java new file mode 100644 index 000000000000..52629d1254e8 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperator.java @@ -0,0 +1,167 @@ +/* + * 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; + +import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.flink.sink.coordinator.CommittableEvent; +import org.apache.paimon.flink.sink.coordinator.CommittingWriteOperatorCoordinator; +import org.apache.paimon.flink.sink.coordinator.WatermarkEvent; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessageSerializer; +import org.apache.paimon.utils.Preconditions; + +import org.apache.flink.api.common.state.ListState; +import org.apache.flink.api.common.state.ListStateDescriptor; +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer; +import org.apache.flink.runtime.operators.coordination.OperatorEventGateway; +import org.apache.flink.runtime.state.StateInitializationContext; +import org.apache.flink.runtime.state.StateSnapshotContext; +import org.apache.flink.streaming.api.operators.StreamOperatorParameters; +import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.TreeMap; + +/** + * Write operator that hands committables to a JM-side {@link CommittingWriteOperatorCoordinator} + * which performs the commit. + * + *

This operator is stateful: it keeps an independent operator state that buffers the + * per-checkpoint committables which have not yet been acknowledged by the coordinator, so they + * survive a global failover and can be replayed on restore. + */ +public class CoordinatorCommittingRowDataStoreWriteOperator + extends StatelessRowDataStoreWriteOperator { + + private static final long serialVersionUID = 1L; + + private static final Logger LOG = + LoggerFactory.getLogger(CoordinatorCommittingRowDataStoreWriteOperator.class); + + @VisibleForTesting + static final String PENDING_COMMITTABLE_STATE_NAME = "pending_committable_state"; + + private final OperatorEventGateway operatorEventGateway; + + /** Persisted buffer of committables not yet acknowledged by the coordinator. */ + private transient ListState pendingCommittableState; + + /** In-memory view of {@link #pendingCommittableState}, grouped by checkpoint id. */ + private transient NavigableMap> pendingCommittables; + + private transient ListSerializer serializer; + + public CoordinatorCommittingRowDataStoreWriteOperator( + StreamOperatorParameters parameters, + FileStoreTable table, + StoreSinkWrite.Provider storeSinkWriteProvider, + String initialCommitUser, + OperatorEventGateway operatorEventGateway) { + super(parameters, table, storeSinkWriteProvider, initialCommitUser); + this.operatorEventGateway = Preconditions.checkNotNull(operatorEventGateway); + } + + @Override + public void initializeState(StateInitializationContext context) throws Exception { + super.initializeState(context); + serializer = CommittableEvent.createCommittableListSerializer(); + pendingCommittableState = + new SimpleVersionedListState<>( + context.getOperatorStateStore() + .getListState( + new ListStateDescriptor<>( + PENDING_COMMITTABLE_STATE_NAME, + BytePrimitiveArraySerializer.INSTANCE)), + new CommittableSerializer(new CommitMessageSerializer())); + pendingCommittables = new TreeMap<>(); + + if (context.isRestored()) { + // Always replay on restore, even if the buffer is empty: the coordinator relies on + // this signal to drive its own restore alignment. + Preconditions.checkState(context.getRestoredCheckpointId().isPresent()); + long checkpointId = context.getRestoredCheckpointId().getAsLong(); + List restored = new ArrayList<>(); + pendingCommittableState.get().forEach(restored::add); + LOG.info("Restore {} of checkpoint {} from state", restored, checkpointId); + pendingCommittableState.clear(); + sendCommittableEventToCoordinator(checkpointId, true, restored); + } + } + + @Override + public void snapshotState(StateSnapshotContext context) throws Exception { + super.snapshotState(context); + pendingCommittableState.clear(); + for (List committables : pendingCommittables.values()) { + pendingCommittableState.addAll(committables); + } + } + + @Override + public void notifyCheckpointComplete(long checkpointId) throws Exception { + super.notifyCheckpointComplete(checkpointId); + // operator state already persisted these; we no longer need to replay them on the next + // restore + Map> completed = pendingCommittables.headMap(checkpointId, true); + completed.clear(); + } + + @Override + protected void emitCommittables(boolean waitCompaction, long checkpointId) throws IOException { + List committables = prepareCommit(waitCompaction, checkpointId); + // send the event regardless of whether the list is empty: the coordinator uses an event + // per (subtask, checkpoint) for alignment + sendCommittableEventToCoordinator(checkpointId, false, committables); + if (!committables.isEmpty()) { + pendingCommittables.put(checkpointId, committables); + } + // Still forward committables downstream even though the commit happens on the coordinator. + // The downstream is a DiscardingSink, but emitting keeps numRecordsOut observable and + // preserves the operator's IO metrics. + committables.forEach(committable -> output.collect(new StreamRecord<>(committable))); + } + + private void sendCommittableEventToCoordinator( + long checkpointId, boolean isRestoring, List committables) + throws IOException { + CommittableEvent event = + CommittableEvent.create(checkpointId, isRestoring, committables, serializer); + operatorEventGateway.sendEventToCoordinator(event); + } + + @Override + public void processWatermark(Watermark mark) throws Exception { + super.processWatermark(mark); + operatorEventGateway.sendEventToCoordinator(new WatermarkEvent(mark)); + } + + @VisibleForTesting + NavigableMap> getPendingCommittables() { + return pendingCommittables; + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java index 9948863c547b..7405ae4894e3 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java @@ -19,6 +19,7 @@ package org.apache.paimon.flink.sink; import org.apache.paimon.CoreOptions.TagCreationMode; +import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.flink.compact.changelog.ChangelogCompactCoordinateOperator; import org.apache.paimon.flink.compact.changelog.ChangelogCompactSortOperator; import org.apache.paimon.flink.compact.changelog.ChangelogCompactWorkerOperator; @@ -26,6 +27,7 @@ import org.apache.paimon.manifest.ManifestCommittable; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; +import org.apache.paimon.table.BucketMode; import org.apache.paimon.table.FileStoreTable; import org.apache.flink.api.common.RuntimeExecutionMode; @@ -50,6 +52,7 @@ import java.util.Queue; import java.util.Set; +import static org.apache.paimon.CoreOptions.WRITE_ONLY; import static org.apache.paimon.CoreOptions.createCommitUser; import static org.apache.paimon.flink.FlinkConnectorOptions.END_INPUT_WATERMARK; import static org.apache.paimon.flink.FlinkConnectorOptions.PRECOMMIT_COMPACT; @@ -187,7 +190,6 @@ public DataStream doWrite( public DataStreamSink doCommit(DataStream written, String commitUser) { StreamExecutionEnvironment env = written.getExecutionEnvironment(); - ReadableConfig conf = env.getConfiguration(); CheckpointConfig checkpointConfig = env.getCheckpointConfig(); boolean streamingCheckpointEnabled = isStreaming(written) && checkpointConfig.isCheckpointingEnabled(); @@ -195,7 +197,32 @@ public DataStreamSink doCommit(DataStream written, String commit assertStreamingConfiguration(env); } + if (coordinatorCommitEnabled()) { + return doCoordinatorCommit(written, checkpointConfig, streamingCheckpointEnabled); + } + return doOperatorCommit(written, commitUser, streamingCheckpointEnabled); + } + + private DataStreamSink doCoordinatorCommit( + DataStream written, + CheckpointConfig checkpointConfig, + boolean streamingCheckpointEnabled) { + checkCoordinatorCommitPreconditions(table, checkpointConfig, streamingCheckpointEnabled); + // The commit runs inside the writer's OperatorCoordinator on the JobManager, so there + // is no global committer operator. Committables are still forwarded by the writer for + // observability and are discarded here. + return written.sinkTo(new PaimonDiscardingSink<>(table)) + .name("end") + .setParallelism(written.getParallelism()); + } + + private DataStreamSink doOperatorCommit( + DataStream written, + String commitUser, + boolean streamingCheckpointEnabled) { + ReadableConfig conf = written.getExecutionEnvironment().getConfiguration(); Options options = Options.fromMap(table.options()); + OneInputStreamOperatorFactory committerOperator = createCommitterOperatorFactory( streamingCheckpointEnabled, commitUser, options.get(END_INPUT_WATERMARK)); @@ -310,6 +337,74 @@ protected abstract OneInputStreamOperatorFactory createWriteOper protected abstract CommittableStateManager createCommittableStateManager(); + /** + * Whether the writer commits through its {@link + * org.apache.flink.runtime.operators.coordination.OperatorCoordinator} on the JobManager. When + * {@code true}, {@link #doCommit} skips the global committer operator. Sinks that wire a + * coordinated writer factory override this. + */ + protected boolean coordinatorCommitEnabled() { + return false; + } + + /** + * Fails fast when coordinator commit is enabled together with a table property or runtime + * setting it cannot honor yet. Enabling coordinator commit is an explicit user choice, so a + * conflicting configuration is reported instead of silently falling back to the global + * committer. + */ + @VisibleForTesting + static void checkCoordinatorCommitPreconditions( + FileStoreTable table, + CheckpointConfig checkpointConfig, + boolean streamingCheckpointEnabled) { + Options options = Options.fromMap(table.options()); + + // Region failover only benefits streaming jobs. A batch job persists its shuffle data and + // has no region-failover concern, so coordinator commit brings no benefit and batch is not + // supported. The commit is driven by checkpoint completion, so checkpointing must be on. + checkArgument( + streamingCheckpointEnabled, + "Could not enable coordinator commit because it requires streaming mode with checkpointing enabled."); + + // The checks below reject configurations that introduce an all-to-all shuffle. Such a + // shuffle places every operator into a single failover region, which defeats the region + // failover that coordinator commit is meant to enable. + checkArgument( + table.primaryKeys().isEmpty(), + "Could not enable coordinator commit because only append tables are supported, but the table has primary keys."); + checkArgument( + table.bucketMode() == BucketMode.BUCKET_UNAWARE, + "Could not enable coordinator commit because bucket mode is " + + table.bucketMode() + + ", only unaware bucket is supported."); + checkArgument( + table.coreOptions().writeOnly(), + "Could not enable coordinator commit because it requires " + + WRITE_ONLY.key() + + " = true."); + checkArgument( + !options.get(PRECOMMIT_COMPACT), + "Could not enable coordinator commit because it requires " + + PRECOMMIT_COMPACT.key() + + " = false."); + + // The OperatorCoordinator cannot tell a savepoint from a normal checkpoint. + // TODO support savepoint auto-tag. + checkArgument( + !options.get(SINK_AUTO_TAG_FOR_SAVEPOINT), + "Could not enable coordinator commit because " + + SINK_AUTO_TAG_FOR_SAVEPOINT.key() + + " is enabled, which is not supported yet."); + + // TODO concurrent checkpoints are not supported yet. + checkArgument( + checkpointConfig.getMaxConcurrentCheckpoints() == 1, + "Could not enable coordinator commit because execution.checkpointing.max-concurrent-checkpoints is " + + checkpointConfig.getMaxConcurrentCheckpoints() + + ", which is not supported yet."); + } + public static boolean isStreaming(DataStream input) { return isStreaming(input.getExecutionEnvironment()); } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PrepareCommitOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PrepareCommitOperator.java index 35f5ff15b9ae..231faa4058fd 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PrepareCommitOperator.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PrepareCommitOperator.java @@ -111,7 +111,7 @@ public void close() throws Exception { } } - private void emitCommittables(boolean waitCompaction, long checkpointId) throws IOException { + protected void emitCommittables(boolean waitCompaction, long checkpointId) throws IOException { prepareCommit(waitCompaction, checkpointId) .forEach(committable -> output.collect(new StreamRecord<>(committable))); } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java index 6e3272f9e0e9..b926dae4c2e5 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java @@ -19,10 +19,19 @@ package org.apache.paimon.flink.sink; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.flink.FlinkConnectorOptions; +import org.apache.paimon.flink.sink.coordinator.CommittingWriteOperatorCoordinator; import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.options.Options; import org.apache.paimon.table.FileStoreTable; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEventGateway; +import org.apache.flink.streaming.api.operators.CoordinatedOperatorFactory; import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory; +import org.apache.flink.streaming.api.operators.StreamOperator; +import org.apache.flink.streaming.api.operators.StreamOperatorParameters; import java.util.Map; @@ -39,11 +48,101 @@ public RowAppendTableSink( @Override protected OneInputStreamOperatorFactory createWriteOperatorFactory( StoreSinkWrite.Provider writeProvider, String commitUser) { + if (coordinatorCommitEnabled()) { + return createCoordinatorCommittingRowWriteOperatorFactory( + table, + writeProvider, + commitUser, + // checkpointing on by default for the JM-side committer; bounded sources will + // be handled by end-input support in a follow-up PR + true, + true, + createCommitterFactory()); + } return createNoStateRowWriteOperatorFactory(table, writeProvider, commitUser); } + @Override + protected boolean coordinatorCommitEnabled() { + return new Options(table.options()) + .get(FlinkConnectorOptions.SINK_COORDINATOR_COMMIT_ENABLED); + } + @Override protected CommittableStateManager createCommittableStateManager() { return createRestoreOnlyCommittableStateManager(table); } + + /** + * Creates a writer factory whose committer runs in a JM-side {@link + * CommittingWriteOperatorCoordinator} instead of a downstream global committer. + */ + private static OneInputStreamOperatorFactory + createCoordinatorCommittingRowWriteOperatorFactory( + FileStoreTable table, + StoreSinkWrite.Provider writeProvider, + String commitUser, + boolean streamingCheckpointEnabled, + boolean failoverAfterRecovery, + Committer.Factory committerFactory) { + return new CoordinatorCommittingFactory( + table, + writeProvider, + commitUser, + streamingCheckpointEnabled, + failoverAfterRecovery, + committerFactory); + } + + private static class CoordinatorCommittingFactory extends RowDataStoreWriteOperator.Factory + implements CoordinatedOperatorFactory { + + private static final long serialVersionUID = 1L; + + private final boolean streamingCheckpointEnabled; + private final boolean failoverAfterRecovery; + private final Committer.Factory committerFactory; + + CoordinatorCommittingFactory( + FileStoreTable table, + StoreSinkWrite.Provider storeSinkWriteProvider, + String initialCommitUser, + boolean streamingCheckpointEnabled, + boolean failoverAfterRecovery, + Committer.Factory committerFactory) { + super(table, storeSinkWriteProvider, initialCommitUser); + this.streamingCheckpointEnabled = streamingCheckpointEnabled; + this.failoverAfterRecovery = failoverAfterRecovery; + this.committerFactory = committerFactory; + } + + @Override + public OperatorCoordinator.Provider getCoordinatorProvider( + String operatorName, OperatorID operatorID) { + return new CommittingWriteOperatorCoordinator.Provider( + operatorID, + committerFactory, + streamingCheckpointEnabled, + initialCommitUser, + failoverAfterRecovery); + } + + @Override + @SuppressWarnings("unchecked") + public > T createStreamOperator( + StreamOperatorParameters parameters) { + OperatorID operatorId = parameters.getStreamConfig().getOperatorID(); + OperatorEventGateway gateway = + parameters.getOperatorEventDispatcher().getOperatorEventGateway(operatorId); + return (T) + new CoordinatorCommittingRowDataStoreWriteOperator( + parameters, table, storeSinkWriteProvider, initialCommitUser, gateway); + } + + @Override + @SuppressWarnings("rawtypes") + public Class getStreamOperatorClass(ClassLoader classLoader) { + return CoordinatorCommittingRowDataStoreWriteOperator.class; + } + } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittableEvent.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittableEvent.java new file mode 100644 index 000000000000..f23474e828d4 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittableEvent.java @@ -0,0 +1,96 @@ +/* + * 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.CommittableSerializer; +import org.apache.paimon.table.sink.CommitMessageSerializer; + +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy; +import org.apache.flink.core.memory.DataOutputSerializer; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; + +import java.io.IOException; +import java.util.List; + +/** + * Operator event sent from a writer subtask to {@link CommittingWriteOperatorCoordinator}, carrying + * the committables produced for one checkpoint. + * + *

{@link Committable} is not directly serializable, so the payload is pre-serialized to {@code + * byte[]} and decoded on the coordinator side. + * + *

{@code isRestoring=true} marks events emitted from {@code initializeState} after a global + * failover, so the coordinator can distinguish replay traffic from normal in-flight checkpoints. + */ +public class CommittableEvent implements OperatorEvent { + + private static final long serialVersionUID = 1L; + + private final long checkpointId; + private final boolean isRestoring; + private final byte[] serialized; + + private CommittableEvent(long checkpointId, boolean isRestoring, byte[] serialized) { + this.checkpointId = checkpointId; + this.isRestoring = isRestoring; + this.serialized = serialized; + } + + public long getCheckpointId() { + return checkpointId; + } + + public boolean isRestoring() { + return isRestoring; + } + + public byte[] getSerialized() { + return serialized; + } + + @Override + public String toString() { + return String.format( + "CommittableEvent{checkpointId=%d, isRestoring=%b, serializedSize=%d}", + checkpointId, isRestoring, serialized.length); + } + + public static CommittableEvent create( + long checkpointId, + boolean isRestoring, + List committables, + ListSerializer listSerializer) + throws IOException { + DataOutputSerializer out = new DataOutputSerializer(256); + listSerializer.serialize(committables, out); + return new CommittableEvent(checkpointId, isRestoring, out.getCopyOfBuffer()); + } + + public static ListSerializer createCommittableListSerializer() { + return new ListSerializer<>(createCommittableSerializer()); + } + + private static TypeSerializer createCommittableSerializer() { + return new SimpleVersionedSerializerTypeSerializerProxy<>( + () -> new CommittableSerializer(new CommitMessageSerializer())); + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java new file mode 100644 index 000000000000..b20348872604 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java @@ -0,0 +1,580 @@ +/* + * 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.annotation.VisibleForTesting; +import org.apache.paimon.flink.sink.Committable; +import org.apache.paimon.flink.sink.Committer; +import org.apache.paimon.flink.sink.state.CoordinatorState; +import org.apache.paimon.flink.sink.state.CoordinatorStateSerializer; +import org.apache.paimon.flink.sink.state.MemoryBackendStateStore; +import org.apache.paimon.manifest.ManifestCommittable; + +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.core.io.SimpleVersionedSerialization; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator; +import org.apache.flink.util.function.ThrowingRunnable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.TreeMap; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import static org.apache.paimon.utils.Preconditions.checkState; + +/** + * {@link OperatorCoordinator} that runs the Paimon committer on the JobManager for the + * unaware-bucket append write path. Writers stream per-checkpoint committables to this coordinator + * over the {@link OperatorEvent} channel; on {@link #notifyCheckpointComplete} the coordinator + * aligns committables across subtasks and commits them from a dedicated single-thread executor, so + * the JM main thread is never blocked by table I/O. + * + *

Wrap this class with a {@link RecreateOnResetOperatorCoordinator} (see {@link Provider}). The + * wrapper discards this instance on global failover and creates a new one in its place, which keeps + * the lifecycle inside a single instance simple. See {@link #resetToCheckpoint} and {@link State} + * for the resulting state machine. + */ +public class CommittingWriteOperatorCoordinator implements OperatorCoordinator { + + private static final Logger LOG = + LoggerFactory.getLogger(CommittingWriteOperatorCoordinator.class); + + private final OperatorCoordinator.Context context; + private final Committer.Factory committerFactory; + private final boolean streamingCheckpointEnabled; + private final boolean failoverAfterRecovery; + private final int parallelism; + + private final WriterCommittables[] subtaskCommittables; + private final ListSerializer committableSerializer; + private final CoordinatorStateSerializer stateSerializer; + private final ExecutorService commitExecutor; + private final SimpleWatermarkValve watermarkValve; + + // Populated by resetToCheckpoint and consumed by start. Plain fields are sufficient: both + // callbacks run on the same scheduler thread in order. + private long restoredCheckpointId = OperatorCoordinator.NO_CHECKPOINT; + private byte[] restoredCheckpointData; + + private State state; + private Committer committer; + private String commitUser; + private MemoryBackendStateStore stateStore; + + public CommittingWriteOperatorCoordinator( + OperatorCoordinator.Context context, + Committer.Factory committerFactory, + boolean streamingCheckpointEnabled, + String initialCommitUser, + boolean failoverAfterRecovery) { + this.context = context; + this.committerFactory = committerFactory; + this.streamingCheckpointEnabled = streamingCheckpointEnabled; + this.commitUser = initialCommitUser; + this.failoverAfterRecovery = failoverAfterRecovery; + this.parallelism = context.currentParallelism(); + this.subtaskCommittables = new WriterCommittables[parallelism]; + this.committableSerializer = CommittableEvent.createCommittableListSerializer(); + this.stateSerializer = new CoordinatorStateSerializer(); + this.commitExecutor = + Executors.newSingleThreadExecutor( + new CoordinatorExecutorThreadFactory("WriteCommitCoordinator", context)); + this.watermarkValve = new SimpleWatermarkValve(parallelism); + this.state = State.CREATED; + } + + @Override + public void start() throws Exception { + // Invoked at most once. If resetToCheckpoint ran first it already moved state to RESTORING + // and stashed the bytes; otherwise we are in CREATED and there is nothing to restore. + checkState( + state == State.CREATED || state == State.RESTORING, + "Coordinator already started, illegal state %s", + state); + runInEventLoop( + () -> { + if (state == State.RESTORING) { + restoreState(restoredCheckpointId, restoredCheckpointData); + // not needed after deserialization; release the reference + restoredCheckpointData = null; + initializeCommitter(true); + // stay in RESTORING until writers re-emit committables and align catches up + } else { + restoreState(OperatorCoordinator.NO_CHECKPOINT, null); + initializeCommitter(false); + transitionState(State.RUNNING); + } + }, + "starting"); + } + + @Override + public void close() throws Exception { + transitionState(State.CLOSED); + if (commitExecutor != null) { + commitExecutor.shutdownNow(); + } + if (committer != null) { + committer.close(); + committer = null; + } + } + + @Override + public void checkpointCoordinator(long checkpointId, CompletableFuture result) { + runCheckpointInEventLoop( + () -> { + if (state != State.RUNNING) { + // if checkpoint is executed before finishing restoring, just fail it + result.completeExceptionally( + new IllegalStateException( + "Checkpoint of commit coordinator should be taken in RUNNING state, while current state is " + + state)); + return; + } + committer.snapshotState(); + byte[] checkpointData = + SimpleVersionedSerialization.writeVersionAndSerialize( + stateSerializer, + new CoordinatorState( + commitUser, + watermarkValve.getCurrentWatermark(), + stateStore.getSerializedStates())); + result.complete(checkpointData); + }, + result, + "taking checkpoint %d", + checkpointId); + } + + @Override + public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event) { + runInEventLoop( + () -> { + if (event instanceof CommittableEvent) { + handleCommittableEvent(subtask, (CommittableEvent) event); + } else if (event instanceof WatermarkEvent) { + handleWatermarkEvent(subtask, (WatermarkEvent) event); + } else { + // TODO: end input handling + throw new UnsupportedOperationException("Unsupported event type: " + event); + } + }, + "handling operator event %s from subtask %d (#%d)", + event, + subtask, + attemptNumber); + } + + @Override + public void notifyCheckpointComplete(long checkpointId) { + runInEventLoop( + () -> { + if (state != State.RUNNING) { + throw new IllegalStateException( + "Completing checkpoint should be notified in RUNNING state, while current state is " + + state); + } + // writers always report a committable per (subtask, checkpoint) during + // snapshot, even if empty; missing means the writer is broken + if (!alignCommittables(checkpointId)) { + throw new IllegalStateException("Not all committables reported by writer"); + } + commitUpToCheckpoint( + checkpointId, + pollManifestCommittablesForCheckpoint( + checkpointId, + subtaskCommittables, + committer, + watermarkValve.getCurrentWatermark()), + committables -> { + try { + committer.commit(committables); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + }, + "completing checkpoint %d", + checkpointId); + } + + /** + * Called by the framework at most once, before {@link #start()}. May be skipped entirely if the + * job has no checkpoint or savepoint to restore from; in that case the coordinator goes + * straight from {@code CREATED} to {@code RUNNING} in {@code start()}. + * + *

When invoked, {@code checkpointId} is the persisted checkpoint id and {@code + * checkpointData} is its bytes; for batch mode, disabled checkpointing, or no completed + * checkpoint, the framework calls it with {@link OperatorCoordinator#NO_CHECKPOINT} and {@code + * null}, which is treated as a no-op. + * + *

The wrapping {@link RecreateOnResetOperatorCoordinator} replaces this instance with a + * fresh one before any further reset, so this method is never called on an already-started + * coordinator. + */ + @Override + public void resetToCheckpoint(long checkpointId, byte[] checkpointData) { + checkState( + state == State.CREATED, + "resetToCheckpoint must run before start, but current state is %s", + state); + if (checkpointId == OperatorCoordinator.NO_CHECKPOINT) { + // nothing to restore; start() will initialize an empty committer + return; + } + restoredCheckpointId = checkpointId; + restoredCheckpointData = checkpointData; + transitionState(State.RESTORING); + } + + @Override + public void subtaskReset(int subtask, long checkpointId) { + runInEventLoop( + () -> { + WriterCommittables writerCommittables = subtaskCommittables[subtask]; + if (writerCommittables != null) { + // sanity check subtask state + Map> committables = + writerCommittables.getCommittablesBeforeCheckpoint( + checkpointId, false); + if (!committables.isEmpty()) { + throw new IllegalStateException( + String.format( + "Writer [%d] contains invalid committables before checkpoint %d", + subtask, checkpointId)); + } + writerCommittables.reset(); + } + }, + "resetting subtask %d to checkpoint %d", + subtask, + checkpointId); + } + + @Override + public void executionAttemptFailed(int subtask, int attemptNumber, Throwable reason) {} + + @Override + public void executionAttemptReady(int subtask, int attemptNumber, SubtaskGateway gateway) {} + + private void handleCommittableEvent(int subtask, CommittableEvent event) throws Exception { + if (state == State.RESTORING) { + updateSubtaskCommittables(subtask, event); + if (alignCommittables(event.getCheckpointId())) { + recover(event.getCheckpointId()); + transitionState(State.RUNNING); + } + } else if (state == State.RUNNING) { + if (event.isRestoring()) { + // a region failover replayed restoring committables while the coordinator itself is + // not restoring; it already holds the committed state, so ignore them + LOG.info( + "Ignore restoring committables from subtask {} of checkpoint {}, coordinator is running.", + subtask, + event.getCheckpointId()); + } else { + updateSubtaskCommittables(subtask, event); + } + } else { + throw new IllegalStateException( + "Illegal state " + state + " while handling committable event " + event); + } + } + + private void updateSubtaskCommittables(int subtask, CommittableEvent event) throws IOException { + WriterCommittables incoming = WriterCommittables.from(event, committableSerializer); + if (subtaskCommittables[subtask] != null) { + subtaskCommittables[subtask].mergeWith(incoming); + } else { + subtaskCommittables[subtask] = incoming; + } + } + + private void handleWatermarkEvent(int subtask, WatermarkEvent event) { + watermarkValve.updateSubtaskWatermark(subtask, event.getWatermark()); + } + + private boolean alignCommittables(long checkpointId) { + for (WriterCommittables committables : subtaskCommittables) { + if (committables == null || committables.getMaxCheckpointId() < checkpointId) { + return false; + } + } + return true; + } + + // replaces CommittableStateManager because committables are not stored in the committer + private void recover(long checkpointId) throws Exception { + if (failoverAfterRecovery) { + // recommit the restored committables and trigger a failover to reinitialize all writers + commitUpToCheckpoint( + checkpointId, + pollManifestCommittablesForCheckpoint( + checkpointId, + subtaskCommittables, + committer, + watermarkValve.getCurrentWatermark()), + committables -> { + int numCommitted = committer.filterAndCommit(committables, true, true); + if (numCommitted > 0) { + throw new RuntimeException( + "This exception is intentionally thrown after committing the " + + "restored checkpoints. By restarting the job we hope " + + "that writers can start writing based on these new commits."); + } + }); + } else { + // just abandon the restoring committables + for (WriterCommittables subtaskCommit : subtaskCommittables) { + subtaskCommit.clearCommittablesBeforeCheckpoint(checkpointId, true); + } + } + } + + @VisibleForTesting + static NavigableMap pollManifestCommittablesForCheckpoint( + long checkpointId, + WriterCommittables[] subtaskCommittables, + Committer committer, + long watermark) + throws IOException { + NavigableMap committablesPerCheckpoint = new TreeMap<>(); + for (WriterCommittables committables : subtaskCommittables) { + NavigableMap> perCheckpoint = + committables.getCommittablesBeforeCheckpoint(checkpointId, true); + for (Map.Entry> entry : perCheckpoint.entrySet()) { + long currentCheckpointId = entry.getKey(); + List currentCommittables = entry.getValue(); + if (currentCommittables.isEmpty()) { + continue; + } + ManifestCommittable manifestCommittable = + committablesPerCheckpoint.get(currentCheckpointId); + if (manifestCommittable == null) { + committablesPerCheckpoint.put( + currentCheckpointId, + committer.combine(currentCheckpointId, watermark, currentCommittables)); + } else { + committer.combine( + currentCheckpointId, + watermark, + manifestCommittable, + currentCommittables); + } + } + committables.clearCommittablesBeforeCheckpoint(checkpointId, true); + } + return committablesPerCheckpoint; + } + + private void commitUpToCheckpoint( + long checkpointId, Map toCommit, CommitAction commitAction) + throws Exception { + List committables = new ArrayList<>(toCommit.values()); + if (committables.isEmpty() && committer.forceCreatingSnapshot()) { + committables = + Collections.singletonList( + committer.combine( + checkpointId, + watermarkValve.getCurrentWatermark(), + Collections.emptyList())); + } + commitAction.accept(committables); + } + + private void restoreState(long checkpointId, byte[] checkpointData) throws IOException { + if (checkpointData == null) { + stateStore = new MemoryBackendStateStore(); + } else { + CoordinatorState coordinatorState = + SimpleVersionedSerialization.readVersionAndDeSerialize( + stateSerializer, checkpointData); + commitUser = coordinatorState.getCommitUser(); + watermarkValve.reset(coordinatorState.getWatermark()); + stateStore = new MemoryBackendStateStore(coordinatorState.getCommitterStates()); + } + } + + private void initializeCommitter(boolean isRestored) { + // Coordinator runs at parallelism 1 (single instance per JobVertex), matching + // CommitterOperator's contract; hardcode parallelism=1 / subtaskIndex=0 + Committer.Context committerContext = + Committer.createContext( + commitUser, + context.metricGroup(), + streamingCheckpointEnabled, + isRestored, + stateStore, + 1, + 0); + committer = committerFactory.create(committerContext); + } + + private void transitionState(State targetState) { + if (state != targetState) { + LOG.info("Transition state from {} to {}", state, targetState); + state = targetState; + } + } + + /** + * Block until every action previously submitted to the single-thread commit executor has + * finished. Tests use this as a fence after firing events into the coordinator. + */ + @VisibleForTesting + public void waitProcessAllActions() throws Exception { + CompletableFuture future = new CompletableFuture<>(); + runInEventLoop(() -> future.complete(null), "waitProcessAllActions"); + future.get(); + } + + @VisibleForTesting + void runInEventLoop( + ThrowingRunnable action, + String actionName, + Object... actionNameFormatParameters) { + commitExecutor.execute( + () -> { + try { + action.run(); + } catch (Throwable t) { + LOG.error( + "Uncaught exception in CommittingWriteOperatorCoordinator while {}. Triggering job failover.", + String.format(actionName, actionNameFormatParameters), + t); + context.failJob(t); + } + }); + } + + /** + * Same as {@link #runInEventLoop} but also completes {@code result} exceptionally on failure, + * so that Flink's checkpoint coordinator can abort the checkpoint immediately instead of + * waiting for the checkpoint timeout. + */ + private void runCheckpointInEventLoop( + ThrowingRunnable action, + CompletableFuture result, + String actionName, + Object... actionNameFormatParameters) { + commitExecutor.execute( + () -> { + try { + action.run(); + } catch (Throwable t) { + LOG.error( + "Uncaught exception in CommittingWriteOperatorCoordinator while {}. Triggering job failover.", + String.format(actionName, actionNameFormatParameters), + t); + result.completeExceptionally(t); + context.failJob(t); + } + }); + } + + @VisibleForTesting + State getCurrentState() { + return state; + } + + @VisibleForTesting + String getCommitUser() { + return commitUser; + } + + /** + * Lifecycle state of the commit coordinator. + * + *

+     *   CREATED ──(resetToCheckpoint, real id)──► RESTORING ──(writers re-emit, align done)──► RUNNING ──► CLOSED
+     *      │                                                                                    ▲
+     *      └────────────────(start, nothing to restore)─────────────────────────────────────────┘
+     * 
+ */ + public enum State { + /** Initial state; resetToCheckpoint may move it to RESTORING before start. */ + CREATED, + + /** + * Restored state has been loaded, but commits are still rejected until every writer subtask + * has re-emitted its pending committables and alignment catches up. + */ + RESTORING, + + /** Accepting checkpoints and commits. */ + RUNNING, + + CLOSED + } + + /** Commits a list of {@link ManifestCommittable}s, possibly throwing checked exceptions. */ + private interface CommitAction { + void accept(List committables) throws Exception; + } + + /** + * Provider that wraps the inner {@link CommittingWriteOperatorCoordinator} in a {@link + * RecreateOnResetOperatorCoordinator}: on global failover the inner is replaced with a fresh + * instance, so the inner never needs to handle "reset after start". + */ + public static class Provider extends RecreateOnResetOperatorCoordinator.Provider { + + private static final long serialVersionUID = 1L; + + private final Committer.Factory committerFactory; + private final boolean streamingCheckpointEnabled; + private final String initialCommitUser; + private final boolean failoverAfterRecovery; + + public Provider( + OperatorID operatorId, + Committer.Factory committerFactory, + boolean streamingCheckpointEnabled, + String initialCommitUser, + boolean failoverAfterRecovery) { + super(operatorId); + this.committerFactory = committerFactory; + this.streamingCheckpointEnabled = streamingCheckpointEnabled; + this.initialCommitUser = initialCommitUser; + this.failoverAfterRecovery = failoverAfterRecovery; + } + + @Override + public OperatorCoordinator getCoordinator(OperatorCoordinator.Context context) { + return new CommittingWriteOperatorCoordinator( + context, + committerFactory, + streamingCheckpointEnabled, + initialCommitUser, + failoverAfterRecovery); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CoordinatorExecutorThreadFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CoordinatorExecutorThreadFactory.java new file mode 100644 index 000000000000..a21387ef8124 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CoordinatorExecutorThreadFactory.java @@ -0,0 +1,92 @@ +/* + * 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.flink.runtime.operators.coordination.OperatorCoordinator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.concurrent.ThreadFactory; + +/** + * Single-thread executor factory used by the {@link CommittingWriteOperatorCoordinator} to run + * commit work off the JM main thread. Any uncaught exception fails the job via {@link + * OperatorCoordinator.Context#failJob(Throwable)}. + * + *

Adapted from {@code + * org.apache.flink.runtime.source.coordinator.SourceCoordinatorProvider.CoordinatorExecutorThreadFactory}. + */ +public class CoordinatorExecutorThreadFactory + implements ThreadFactory, Thread.UncaughtExceptionHandler { + + private static final Logger LOG = + LoggerFactory.getLogger(CoordinatorExecutorThreadFactory.class); + + private final String coordinatorThreadName; + private final ClassLoader contextClassLoader; + private final Thread.UncaughtExceptionHandler errorHandler; + + @Nullable private Thread coordinatorThread; + + public CoordinatorExecutorThreadFactory( + String coordinatorThreadName, OperatorCoordinator.Context context) { + this( + coordinatorThreadName, + context.getUserCodeClassloader(), + (t, e) -> { + LOG.error( + "Thread '{}' produced an uncaught exception. Failing the job.", + t.getName(), + e); + context.failJob(e); + }); + } + + CoordinatorExecutorThreadFactory( + String coordinatorThreadName, + ClassLoader contextClassLoader, + Thread.UncaughtExceptionHandler errorHandler) { + this.coordinatorThreadName = coordinatorThreadName; + this.contextClassLoader = contextClassLoader; + this.errorHandler = errorHandler; + } + + @Override + public synchronized Thread newThread(Runnable r) { + coordinatorThread = new Thread(r, coordinatorThreadName); + coordinatorThread.setContextClassLoader(contextClassLoader); + coordinatorThread.setUncaughtExceptionHandler(this); + return coordinatorThread; + } + + @Override + public synchronized void uncaughtException(Thread t, Throwable e) { + errorHandler.uncaughtException(t, e); + } + + public String getCoordinatorThreadName() { + return coordinatorThreadName; + } + + public boolean isCurrentThreadCoordinatorThread() { + return Thread.currentThread() == coordinatorThread; + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SimpleWatermarkValve.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SimpleWatermarkValve.java new file mode 100644 index 000000000000..a1b675b1dac8 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SimpleWatermarkValve.java @@ -0,0 +1,110 @@ +/* + * 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.flink.runtime.state.heap.HeapPriorityQueue; +import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Objects; + +/** + * A simple implementation of coordinator watermark alignment. It cannot fully match the operator + * watermark semantics as it lacks idle information. + * + *

And currently, it does not support revert watermark. The watermark always grows. + */ +public class SimpleWatermarkValve { + private static final Logger LOG = LoggerFactory.getLogger(SimpleWatermarkValve.class); + + private final SimpleWatermark[] subtaskWatermarks; + + // reuse HeapPriorityQueue of flink runtime + private final HeapPriorityQueue watermarkPriorityQueue; + + private long currentWatermark; + + public SimpleWatermarkValve(int parallelism) { + this.subtaskWatermarks = new SimpleWatermark[parallelism]; + this.watermarkPriorityQueue = + new HeapPriorityQueue<>( + (left, right) -> + Long.compare(left.getWatermarkValue(), right.getWatermarkValue()), + parallelism); + for (int i = 0; i < parallelism; i++) { + subtaskWatermarks[i] = new SimpleWatermark(); + watermarkPriorityQueue.add(subtaskWatermarks[i]); + } + this.currentWatermark = Long.MIN_VALUE; + } + + public void reset(long watermark) { + LOG.info("Reset watermark valve to {}", watermark); + for (SimpleWatermark simpleWatermark : subtaskWatermarks) { + simpleWatermark.setWatermarkValue(watermark); + } + this.currentWatermark = watermark; + } + + public void updateSubtaskWatermark(int subtask, long watermark) { + if (watermark == Long.MAX_VALUE) { + // Do not consume Long.MAX_VALUE watermark in case of batch or bounded stream + return; + } + if (watermark > subtaskWatermarks[subtask].getWatermarkValue()) { + subtaskWatermarks[subtask].setWatermarkValue(watermark); + watermarkPriorityQueue.adjustModifiedElement(subtaskWatermarks[subtask]); + long minWatermark = + Objects.requireNonNull(watermarkPriorityQueue.peek()).getWatermarkValue(); + if (currentWatermark != minWatermark) { + LOG.debug("Update watermark to {}", minWatermark); + currentWatermark = minWatermark; + } + } + } + + public long getCurrentWatermark() { + return currentWatermark; + } + + private static class SimpleWatermark implements HeapPriorityQueueElement { + private int heapIndex = HeapPriorityQueueElement.NOT_CONTAINED; + + private long watermarkValue = Long.MIN_VALUE; + + public void setWatermarkValue(long watermarkValue) { + this.watermarkValue = watermarkValue; + } + + public long getWatermarkValue() { + return watermarkValue; + } + + @Override + public int getInternalIndex() { + return heapIndex; + } + + @Override + public void setInternalIndex(int newIndex) { + heapIndex = newIndex; + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WatermarkEvent.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WatermarkEvent.java new file mode 100644 index 000000000000..af62e4995bac --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WatermarkEvent.java @@ -0,0 +1,63 @@ +/* + * 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.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.streaming.api.watermark.Watermark; + +import java.util.Objects; + +/** The watermark event sent by writer to coordinator committer through flink rpc. */ +public class WatermarkEvent implements OperatorEvent { + + private static final long serialVersionUID = 1L; + + private final long watermark; + + public WatermarkEvent(Watermark watermark) { + this(watermark.getTimestamp()); + } + + public WatermarkEvent(long watermark) { + this.watermark = watermark; + } + + public long getWatermark() { + return watermark; + } + + @Override + public boolean equals(Object o) { + if (o == null || getClass() != o.getClass()) { + return false; + } + WatermarkEvent that = (WatermarkEvent) o; + return watermark == that.watermark; + } + + @Override + public int hashCode() { + return Objects.hashCode(watermark); + } + + @Override + public String toString() { + return "WatermarkEvent {" + "watermark = " + watermark + "}"; + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WriterCommittables.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WriterCommittables.java new file mode 100644 index 000000000000..f9d8c9e82f64 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/WriterCommittables.java @@ -0,0 +1,137 @@ +/* + * 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.annotation.VisibleForTesting; +import org.apache.paimon.flink.sink.Committable; + +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.TreeMap; + +/** + * Buffer of {@link Committable}s received from one writer subtask, indexed by checkpoint id. Lives + * inside the coordinator and is consumed when a checkpoint is ready to commit. + */ +public class WriterCommittables { + + private static final Logger LOG = LoggerFactory.getLogger(WriterCommittables.class); + + private long maxCheckpointId; + private final NavigableMap> committablesPerCheckpoint; + + @VisibleForTesting + WriterCommittables(long maxCheckpointId, List committables) { + this.maxCheckpointId = maxCheckpointId; + this.committablesPerCheckpoint = new TreeMap<>(); + + if (!committables.isEmpty()) { + groupByCheckpoint(committablesPerCheckpoint, committables); + if (maxCheckpointId < committablesPerCheckpoint.lastKey()) { + throw new IllegalStateException( + "Invalid input committables, max checkpoint id should not be less than " + + "checkpoint id of committables, max checkpoint is " + + maxCheckpointId + + ", committables are " + + committables); + } + } + } + + public NavigableMap> getCommittablesPerCheckpoint() { + return committablesPerCheckpoint; + } + + public NavigableMap> getCommittablesBeforeCheckpoint( + long checkpointId, boolean inclusive) { + return committablesPerCheckpoint.headMap(checkpointId, inclusive); + } + + public void clearCommittablesBeforeCheckpoint(long checkpointId, boolean inclusive) { + if (checkpointId > maxCheckpointId || (checkpointId == maxCheckpointId && inclusive)) { + reset(); + } else { + committablesPerCheckpoint.headMap(checkpointId, inclusive).clear(); + } + } + + public void reset() { + maxCheckpointId = -1; + committablesPerCheckpoint.clear(); + } + + public void mergeWith(WriterCommittables other) { + if (other.maxCheckpointId <= maxCheckpointId) { + throw new IllegalStateException( + "It must merge later checkpoint committables, however current checkpoint id is " + + maxCheckpointId + + ", to be merged checkpoint id is " + + other.maxCheckpointId); + } + maxCheckpointId = other.maxCheckpointId; + for (Map.Entry> entry : + other.getCommittablesPerCheckpoint().entrySet()) { + if (committablesPerCheckpoint.containsKey(entry.getKey())) { + LOG.error( + "Subtask committables should not contain {}, the current committables are " + + "{}, to be merged committables are {}", + entry.getKey(), + committablesPerCheckpoint, + other.getCommittablesPerCheckpoint()); + throw new IllegalStateException( + "Subtask contains repeated committables of checkpoint " + entry.getKey()); + } + committablesPerCheckpoint.put(entry.getKey(), entry.getValue()); + } + } + + public long getMaxCheckpointId() { + return maxCheckpointId; + } + + @Override + public String toString() { + return String.format( + "WriterCommittables{maxCheckpointId=%d, committables=%s}", + maxCheckpointId, committablesPerCheckpoint); + } + + public static WriterCommittables from( + CommittableEvent event, ListSerializer serializer) throws IOException { + DataInputDeserializer in = new DataInputDeserializer(event.getSerialized()); + return new WriterCommittables(event.getCheckpointId(), serializer.deserialize(in)); + } + + @VisibleForTesting + static void groupByCheckpoint( + Map> grouped, Collection committables) { + for (Committable c : committables) { + grouped.computeIfAbsent(c.checkpointId(), k -> new ArrayList<>()).add(c); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/state/CoordinatorState.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/state/CoordinatorState.java new file mode 100644 index 000000000000..a4c656668076 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/state/CoordinatorState.java @@ -0,0 +1,64 @@ +/* + * 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.state; + +import javax.annotation.Nonnull; + +import java.util.Map; + +/** + * The state persisted by {@link + * org.apache.paimon.flink.sink.coordinator.CommittingWriteOperatorCoordinator} during {@code + * checkpointCoordinator}. + * + *

Note that uncommitted committables are not persisted here. Writers keep their own + * uncommitted committables in operator state and re-emit them to the coordinator on {@code + * initializeState}; this avoids races caused by the fact that {@code + * OperatorCoordinator.checkpointCoordinator} runs before any operator subtask has captured its own + * snapshot. + */ +public class CoordinatorState { + + @Nonnull private final String commitUser; + private final long watermark; + @Nonnull private final Map committerStates; + + public CoordinatorState( + @Nonnull String commitUser, + long watermark, + @Nonnull Map committerStates) { + this.commitUser = commitUser; + this.watermark = watermark; + this.committerStates = committerStates; + } + + @Nonnull + public String getCommitUser() { + return commitUser; + } + + public long getWatermark() { + return watermark; + } + + @Nonnull + public Map getCommitterStates() { + return committerStates; + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/state/CoordinatorStateSerializer.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/state/CoordinatorStateSerializer.java new file mode 100644 index 000000000000..4ab88bca7a3f --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/state/CoordinatorStateSerializer.java @@ -0,0 +1,69 @@ +/* + * 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.state; + +import org.apache.paimon.utils.Preconditions; + +import org.apache.flink.api.common.typeutils.base.MapSerializer; +import org.apache.flink.api.common.typeutils.base.StringSerializer; +import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer; +import org.apache.flink.core.io.SimpleVersionedSerializer; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataOutputSerializer; + +import java.io.IOException; +import java.util.Map; + +/** Versioned serializer for {@link CoordinatorState}. */ +public class CoordinatorStateSerializer implements SimpleVersionedSerializer { + + private static final int CURRENT_VERSION = 1; + + private final MapSerializer committerStateSerializer = + new MapSerializer<>(StringSerializer.INSTANCE, BytePrimitiveArraySerializer.INSTANCE); + + @Override + public int getVersion() { + return CURRENT_VERSION; + } + + @Override + public byte[] serialize(CoordinatorState state) throws IOException { + DataOutputSerializer out = new DataOutputSerializer(256); + out.writeUTF(state.getCommitUser()); + out.writeLong(state.getWatermark()); + committerStateSerializer.serialize(state.getCommitterStates(), out); + return out.getCopyOfBuffer(); + } + + @Override + public CoordinatorState deserialize(int version, byte[] serialized) throws IOException { + Preconditions.checkState( + version == CURRENT_VERSION, + "Could not deserialize coordinator state of version " + + version + + ", expected version " + + CURRENT_VERSION); + DataInputDeserializer in = new DataInputDeserializer(serialized); + String commitUser = in.readUTF(); + long watermark = in.readLong(); + Map committerStates = committerStateSerializer.deserialize(in); + return new CoordinatorState(commitUser, watermark, committerStates); + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/state/MemoryBackendStateStore.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/state/MemoryBackendStateStore.java new file mode 100644 index 000000000000..79cddf4b92d8 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/state/MemoryBackendStateStore.java @@ -0,0 +1,131 @@ +/* + * 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.state; + +import org.apache.flink.api.common.state.ListState; +import org.apache.flink.api.common.state.ListStateDescriptor; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataOutputSerializer; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * A {@link StateStore} that lives entirely in memory. Designed to be used inside an {@link + * org.apache.flink.runtime.operators.coordination.OperatorCoordinator}, where state is serialized + * to bytes during {@code checkpointCoordinator} and restored from bytes during {@code + * resetToCheckpoint}. + */ +public class MemoryBackendStateStore implements StateStore { + + private final Map> registeredStates = new HashMap<>(); + private final Map serializedStates; + + public MemoryBackendStateStore() { + this(Collections.emptyMap()); + } + + public MemoryBackendStateStore(Map restoredStates) { + this.serializedStates = new HashMap<>(restoredStates); + } + + @Override + public ListState getListState(ListStateDescriptor descriptor) throws Exception { + String stateName = descriptor.getName(); + MemoryListState existing = registeredStates.get(stateName); + if (existing != null) { + //noinspection unchecked + return (ListState) existing; + } + byte[] restored = serializedStates.get(stateName); + MemoryListState state = + new MemoryListState<>(descriptor.getElementSerializer(), restored); + registeredStates.put(stateName, state); + return state; + } + + /** + * Serializes all currently-registered states to bytes. Restored-but-unread state entries are + * kept verbatim so that they survive coordinator failover even if no consumer registered a + * matching descriptor in this lifecycle. + */ + public Map getSerializedStates() throws IOException { + for (Map.Entry> entry : registeredStates.entrySet()) { + serializedStates.put(entry.getKey(), entry.getValue().serialize()); + } + return serializedStates; + } + + private static class MemoryListState implements ListState { + + private final ListSerializer serializer; + private final List values; + + MemoryListState(TypeSerializer elementSerializer, @Nullable byte[] restored) + throws IOException { + this.serializer = new ListSerializer<>(elementSerializer); + if (restored != null) { + DataInputDeserializer in = new DataInputDeserializer(restored); + this.values = serializer.deserialize(in); + } else { + this.values = new ArrayList<>(); + } + } + + @Override + public Iterable get() { + return values; + } + + @Override + public void add(T value) { + values.add(value); + } + + @Override + public void update(List newValues) { + values.clear(); + values.addAll(newValues); + } + + @Override + public void addAll(List toAdd) { + values.addAll(toAdd); + } + + @Override + public void clear() { + values.clear(); + } + + byte[] serialize() throws IOException { + DataOutputSerializer out = new DataOutputSerializer(256); + serializer.serialize(values, out); + return out.getCopyOfBuffer(); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java new file mode 100644 index 000000000000..0ddbfbd18c48 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CoordinatorCommitITCase.java @@ -0,0 +1,306 @@ +/* + * 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; + +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.reader.RecordReaderIterator; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.utils.CloseableIterator; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.client.program.ClusterClient; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.metrics.Metric; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.groups.OperatorMetricGroup; +import org.apache.flink.runtime.client.JobStatusMessage; +import org.apache.flink.runtime.testutils.InMemoryReporter; +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.TableResult; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +/** End-to-end tests for JM-side commit on unaware-bucket append tables. */ +public class CoordinatorCommitITCase { + + private static final int DEFAULT_PARALLELISM = 2; + private static final long WAIT_TIMEOUT_MILLIS = 60_000L; + private static final InMemoryReporter reporter = InMemoryReporter.create(); + + @RegisterExtension + protected static final org.apache.paimon.flink.util.MiniClusterWithClientExtension + MINI_CLUSTER_EXTENSION = + new org.apache.paimon.flink.util.MiniClusterWithClientExtension( + new MiniClusterResourceConfiguration.Builder() + .setNumberTaskManagers(1) + .setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM) + .setConfiguration( + reporter.addToConfiguration(new Configuration())) + .build()); + + @TempDir Path tempPath; + + @AfterEach + public final void cleanupRunningJobs() throws Exception { + ClusterClient clusterClient = MINI_CLUSTER_EXTENSION.createRestClusterClient(); + for (JobStatusMessage job : clusterClient.listJobs().get()) { + if (!job.getJobState().isTerminalState()) { + try { + clusterClient.cancel(job.getJobId()).get(30, TimeUnit.SECONDS); + } catch (Exception ignored) { + // best-effort cleanup + } + } + } + } + + @Timeout(value = 120, unit = TimeUnit.SECONDS) + @Test + public void testCoordinatorCommitRemovesGlobalCommitterOperator() throws Exception { + RunningJob coordinatorCommitJob = startStreamingInsert(true); + waitUntilWriterMetricGroupsRegistered(coordinatorCommitJob.jobId); + assertThat(findGlobalCommitterMetricGroups(coordinatorCommitJob.jobId)).isEmpty(); + coordinatorCommitJob.cancel(); + + RunningJob defaultCommitJob = startStreamingInsert(false); + waitUntilGlobalCommitterMetricGroupsRegistered(defaultCommitJob.jobId); + defaultCommitJob.cancel(); + } + + @Timeout(value = 120, unit = TimeUnit.SECONDS) + @Test + public void testCoordinatorCommitMetricsAndCommittedRows() throws Exception { + RunningJob runningJob = startStreamingInsert(true); + waitUntilWriterInputRecords(runningJob.jobId); + waitUntilCoordinatorCommitMetricsRegistered(runningJob.jobId); + assertThat(findGlobalCommitterMetricGroups(runningJob.jobId)).isEmpty(); + waitUntilRowsCommitted(runningJob); + runningJob.cancel(); + + assertThat(readRowCount(runningJob.table)).isGreaterThan(0L); + } + + private RunningJob startStreamingInsert(boolean coordinatorCommitEnabled) throws Exception { + String tableName = coordinatorCommitEnabled ? "T_COORDINATOR_COMMIT" : "T_DEFAULT_COMMIT"; + TableEnvironment tEnv = + TableEnvironment.create( + EnvironmentSettings.newInstance().inStreamingMode().build()); + tEnv.getConfig().getConfiguration().setString("execution.checkpointing.interval", "200 ms"); + + tEnv.executeSql( + "CREATE CATALOG mycat WITH ( 'type' = 'paimon', 'warehouse' = '" + + tempPath + + "' )"); + tEnv.executeSql("USE CATALOG mycat"); + String coordinatorCommitOption = + coordinatorCommitEnabled + ? ", 'sink.coordinator-commit.enabled' = 'true', 'write-only' = 'true'" + : ""; + tEnv.executeSql( + "CREATE TABLE " + + tableName + + " (id INT, data STRING) WITH (" + + "'bucket' = '-1'" + + coordinatorCommitOption + + ")"); + tEnv.executeSql( + "CREATE TEMPORARY TABLE src (id INT, data STRING) WITH (" + + "'connector' = 'datagen', " + + "'rows-per-second' = '20')"); + TableResult tableResult = + tEnv.executeSql("INSERT INTO " + tableName + " SELECT * FROM src"); + JobClient client = tableResult.getJobClient().get(); + FileStoreTable table = + (FileStoreTable) + ((FlinkCatalog) tEnv.getCatalog("mycat").get()) + .catalog() + .getTable(Identifier.create("default", tableName)); + return new RunningJob(table, client); + } + + private List findGlobalCommitterMetricGroups(JobID jobId) { + return reporter.findOperatorMetricGroups(jobId, "Global Committer.*"); + } + + private void waitUntilWriterMetricGroupsRegistered(JobID jobId) throws Exception { + long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS; + while (reporter.findOperatorMetricGroups(jobId, "Writer.*").isEmpty() + && System.currentTimeMillis() < deadline) { + Thread.sleep(500); + } + assertThat(reporter.findOperatorMetricGroups(jobId, "Writer.*")).isNotEmpty(); + } + + private void waitUntilGlobalCommitterMetricGroupsRegistered(JobID jobId) throws Exception { + long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS; + while (findGlobalCommitterMetricGroups(jobId).isEmpty() + && System.currentTimeMillis() < deadline) { + Thread.sleep(500); + } + assertThat(findGlobalCommitterMetricGroups(jobId)).isNotEmpty(); + } + + private void waitUntilWriterInputRecords(JobID jobId) throws Exception { + long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS; + while (System.currentTimeMillis() < deadline) { + if (writerInputRecords(jobId) > 0) { + return; + } + Thread.sleep(500); + } + assertThat(writerInputRecords(jobId)) + .describedAs(describeWriterMetrics(jobId)) + .isPositive(); + } + + private long writerInputRecords(JobID jobId) { + long records = 0L; + for (OperatorMetricGroup group : reporter.findOperatorMetricGroups(jobId, "Writer.*")) { + records += group.getIOMetricGroup().getNumRecordsInCounter().getCount(); + } + return records; + } + + private void waitUntilCoordinatorCommitMetricsRegistered(JobID jobId) throws Exception { + long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS; + while (System.currentTimeMillis() < deadline) { + if (hasCoordinatorCommitMetrics(jobId)) { + return; + } + Thread.sleep(500); + } + assertThat(hasCoordinatorCommitMetrics(jobId)) + .describedAs(describeCommitMetrics(jobId)) + .isTrue(); + } + + private boolean hasCoordinatorCommitMetrics(JobID jobId) { + for (java.util.Map.Entry> group : + reporter.getMetricsByGroup().entrySet()) { + if (!isMetricGroupForJob(group.getKey(), jobId)) { + continue; + } + if (!group.getKey().getClass().getSimpleName().equals("GenericMetricGroup")) { + continue; + } + for (String metricName : group.getValue().keySet()) { + if (metricName.equals("commitDuration") + || metricName.equals("lastCommittedSnapshotId")) { + return true; + } + } + } + return false; + } + + private String describeCommitMetrics(JobID jobId) { + StringBuilder builder = new StringBuilder("commit metrics:"); + for (java.util.Map.Entry> group : + reporter.getMetricsByGroup().entrySet()) { + if (!isMetricGroupForJob(group.getKey(), jobId)) { + continue; + } + for (java.util.Map.Entry metric : group.getValue().entrySet()) { + if (metric.getKey().contains("Commit") + || metric.getKey().contains("commit") + || metric.getKey().contains("numRecordsOut")) { + builder.append(' ') + .append(group.getKey().getClass().getSimpleName()) + .append('#') + .append(metric.getKey()) + .append('=') + .append(metric.getValue().getClass().getSimpleName()); + } + } + } + return builder.toString(); + } + + private boolean isMetricGroupForJob(MetricGroup metricGroup, JobID jobId) { + return metricGroup.getAllVariables().containsValue(jobId.toString()); + } + + private String describeWriterMetrics(JobID jobId) { + StringBuilder builder = new StringBuilder("writer metrics:"); + for (OperatorMetricGroup group : reporter.findOperatorMetricGroups(jobId, "Writer.*")) { + builder.append(' ') + .append(group.getIOMetricGroup().getNumRecordsInCounter().getCount()) + .append('/') + .append(group.getIOMetricGroup().getNumRecordsOutCounter().getCount()); + } + return builder.toString(); + } + + private long readRowCount(FileStoreTable table) throws Exception { + RecordReader reader = + table.newRead().createReader(table.newSnapshotReader().read()); + long rowCount = 0L; + try (CloseableIterator iterator = new RecordReaderIterator<>(reader)) { + while (iterator.hasNext()) { + iterator.next(); + rowCount++; + } + } + return rowCount; + } + + private void waitUntilRowsCommitted(RunningJob runningJob) throws Exception { + long deadline = System.currentTimeMillis() + WAIT_TIMEOUT_MILLIS; + while (System.currentTimeMillis() < deadline) { + if (readRowCount(runningJob.table) > 0) { + return; + } + Thread.sleep(500); + } + assertThat(readRowCount(runningJob.table)) + .describedAs(describeWriterMetrics(runningJob.jobId)) + .isGreaterThan(0L); + } + + private static class RunningJob { + + private final FileStoreTable table; + private final JobClient client; + private final JobID jobId; + + private RunningJob(FileStoreTable table, JobClient client) { + this.table = table; + this.client = client; + this.jobId = client.getJobID(); + } + + private void cancel() throws Exception { + client.cancel().get(30, TimeUnit.SECONDS); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperatorTest.java new file mode 100644 index 000000000000..fd83749f1ecb --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperatorTest.java @@ -0,0 +1,368 @@ +/* + * 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; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.flink.sink.coordinator.CommittableEvent; +import org.apache.paimon.flink.sink.coordinator.CommittingWriteOperatorCoordinator; +import org.apache.paimon.flink.sink.coordinator.WriterCommittables; +import org.apache.paimon.flink.utils.InternalRowTypeSerializer; +import org.apache.paimon.flink.utils.InternalTypeInfo; +import org.apache.paimon.table.FileStoreTable; + +import org.apache.flink.api.common.ExecutionConfig; +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.runtime.checkpoint.CheckpointCoordinator; +import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.operators.coordination.CoordinatorStore; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.runtime.operators.coordination.OperatorEventGateway; +import org.apache.flink.streaming.api.operators.StreamOperator; +import org.apache.flink.streaming.api.operators.StreamOperatorParameters; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness; +import org.apache.flink.util.FlinkRuntimeException; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link CoordinatorCommittingRowDataStoreWriteOperator}. */ +public class CoordinatorCommittingRowDataStoreWriteOperatorTest extends CommitterOperatorTestBase { + + private static final ListSerializer COMMITTABLE_LIST_SERIALIZER = + CommittableEvent.createCommittableListSerializer(); + + @Test + @Timeout(30) + public void testWriterSendsCommittablesToCoordinatorAndStillEmitsDownstream() throws Exception { + FileStoreTable table = + createFileStoreTable( + options -> { + options.set(CoreOptions.BUCKET, -1); + options.remove("bucket-key"); + }); + String commitUser = UUID.randomUUID().toString(); + List events = new ArrayList<>(); + + CommittingWriteOperatorCoordinator coordinator = + new CommittingWriteOperatorCoordinator( + new TestingContext(), + context -> + new StoreCommitter( + table, table.newCommit(context.commitUser()), context), + true, + commitUser, + false); + coordinator.start(); + coordinator.waitProcessAllActions(); + + OneInputStreamOperatorTestHarness harness = + createHarness(table, commitUser, events::add); + TypeSerializer committableSerializer = + new CommittableTypeInfo().createSerializer(new ExecutionConfig()); + harness.setup(committableSerializer); + harness.open(); + + harness.processElement(GenericRow.of(1, 1L), 1); + harness.prepareSnapshotPreBarrier(1); + harness.snapshot(1, 10); + + List downstreamCommittables = extractCommittables(harness); + assertThat(downstreamCommittables).hasSize(1); + assertThat(events).hasSize(1); + + CommittableEvent event = (CommittableEvent) events.get(0); + assertThat(event.getCheckpointId()).isEqualTo(1L); + assertThat(event.isRestoring()).isFalse(); + assertThat( + WriterCommittables.from(event, COMMITTABLE_LIST_SERIALIZER) + .getCommittablesPerCheckpoint() + .get(1L)) + .hasSize(1); + + coordinator.handleEventFromOperator(0, 0, event); + coordinator.notifyCheckpointComplete(1); + coordinator.waitProcessAllActions(); + harness.notifyOfCompletedCheckpoint(1); + + assertResults(table, "1, 1"); + + harness.close(); + coordinator.close(); + } + + @Test + public void + testPendingCommittablesAccumulateAcrossUncompletedCheckpointsAndAreClearedOnComplete() + throws Exception { + FileStoreTable table = createUnawareBucketTable(); + String commitUser = UUID.randomUUID().toString(); + List events = new ArrayList<>(); + + OneInputStreamOperatorTestHarness harness = + createHarness(table, commitUser, events::add); + TypeSerializer committableSerializer = + new CommittableTypeInfo().createSerializer(new ExecutionConfig()); + harness.setup(committableSerializer); + harness.open(); + CoordinatorCommittingRowDataStoreWriteOperator operator = + (CoordinatorCommittingRowDataStoreWriteOperator) harness.getOperator(); + + // cp1: write data, snapshot, but checkpoint never completes (e.g. aborted) + harness.processElement(GenericRow.of(1, 10L), 1); + harness.prepareSnapshotPreBarrier(1); + harness.snapshot(1, 10); + assertEventCheckpointAndRestoring(events.get(events.size() - 1), 1L, false); + assertThat(operator.getPendingCommittables()).containsKey(1L); + + // cp2: another writing checkpoint, also aborted; cp1 buffer must still be retained + harness.processElement(GenericRow.of(2, 20L), 2); + harness.prepareSnapshotPreBarrier(2); + harness.snapshot(2, 20); + assertEventCheckpointAndRestoring(events.get(events.size() - 1), 2L, false); + assertThat(operator.getPendingCommittables()).containsKeys(1L, 2L); + + // cp3 completes -> headMap(3, true) clears every buffered checkpoint + harness.processElement(GenericRow.of(3, 30L), 3); + harness.prepareSnapshotPreBarrier(3); + harness.snapshot(3, 30); + assertEventCheckpointAndRestoring(events.get(events.size() - 1), 3L, false); + harness.notifyOfCompletedCheckpoint(3); + assertThat(operator.getPendingCommittables()).isEmpty(); + + harness.close(); + } + + @Test + public void testRestoreReplaysBufferedCommittablesToCoordinator() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + String commitUser = UUID.randomUUID().toString(); + TypeSerializer committableSerializer = + new CommittableTypeInfo().createSerializer(new ExecutionConfig()); + + // session 1: write + snapshot, do NOT notify completion, then crash + List firstEvents = new ArrayList<>(); + OneInputStreamOperatorTestHarness firstHarness = + createHarness(table, commitUser, firstEvents::add); + firstHarness.setup(committableSerializer); + firstHarness.open(); + firstHarness.processElement(GenericRow.of(1, 10L), 1); + firstHarness.prepareSnapshotPreBarrier(1); + OperatorSubtaskState snapshot = firstHarness.snapshot(1, 10); + firstHarness.close(); + + // session 2: restore from snapshot, expect a single isRestoring=true event carrying the + // buffered committable, even though the writer has not produced anything new yet + List restoredEvents = new ArrayList<>(); + OneInputStreamOperatorTestHarness secondHarness = + createHarness(table, commitUser, restoredEvents::add); + secondHarness.setup(committableSerializer); + secondHarness.initializeState(snapshot); + secondHarness.open(); + + assertThat(restoredEvents).hasSize(1); + CommittableEvent restoredEvent = (CommittableEvent) restoredEvents.get(0); + assertThat(restoredEvent.isRestoring()).isTrue(); + assertThat(decodeCommittables(restoredEvent)).hasSize(1); + + // a fresh snapshot must persist nothing (the buffer was cleared on restore) + CoordinatorCommittingRowDataStoreWriteOperator operator = + (CoordinatorCommittingRowDataStoreWriteOperator) secondHarness.getOperator(); + secondHarness.prepareSnapshotPreBarrier(2); + secondHarness.snapshot(2, 20); + assertThat(operator.getPendingCommittables()).isEmpty(); + + secondHarness.close(); + } + + @Test + public void testEmptyRestoreStillSendsIsRestoringSignal() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + String commitUser = UUID.randomUUID().toString(); + TypeSerializer committableSerializer = + new CommittableTypeInfo().createSerializer(new ExecutionConfig()); + + // session 1: snapshot with no data, then crash + List firstEvents = new ArrayList<>(); + OneInputStreamOperatorTestHarness firstHarness = + createHarness(table, commitUser, firstEvents::add); + firstHarness.setup(committableSerializer); + firstHarness.open(); + firstHarness.prepareSnapshotPreBarrier(1); + OperatorSubtaskState snapshot = firstHarness.snapshot(1, 10); + firstHarness.close(); + + // session 2: restore. Buffer is empty, but the operator must still notify the coordinator + // so the coordinator can drive its own restore-alignment. + List restoredEvents = new ArrayList<>(); + OneInputStreamOperatorTestHarness secondHarness = + createHarness(table, commitUser, restoredEvents::add); + secondHarness.setup(committableSerializer); + secondHarness.initializeState(snapshot); + secondHarness.open(); + + assertThat(restoredEvents).hasSize(1); + CommittableEvent restoredEvent = (CommittableEvent) restoredEvents.get(0); + assertThat(restoredEvent.isRestoring()).isTrue(); + assertThat(decodeCommittables(restoredEvent)).isEmpty(); + + secondHarness.close(); + } + + private void assertEventCheckpointAndRestoring( + OperatorEvent event, long expectedCheckpointId, boolean expectedRestoring) { + CommittableEvent committableEvent = (CommittableEvent) event; + assertThat(committableEvent.getCheckpointId()).isEqualTo(expectedCheckpointId); + assertThat(committableEvent.isRestoring()).isEqualTo(expectedRestoring); + } + + private List decodeCommittables(CommittableEvent event) throws Exception { + return COMMITTABLE_LIST_SERIALIZER.deserialize( + new DataInputDeserializer(event.getSerialized())); + } + + private FileStoreTable createUnawareBucketTable() throws Exception { + return createFileStoreTable( + options -> { + options.set(CoreOptions.BUCKET, -1); + options.remove("bucket-key"); + }); + } + + @SuppressWarnings("unchecked") + private List extractCommittables( + OneInputStreamOperatorTestHarness harness) { + List committables = new ArrayList<>(); + while (!harness.getOutput().isEmpty()) { + committables.add(((StreamRecord) harness.getOutput().poll()).getValue()); + } + return committables; + } + + private OneInputStreamOperatorTestHarness createHarness( + FileStoreTable table, String commitUser, OperatorEventGateway gateway) + throws Exception { + RowDataStoreWriteOperator.Factory operatorFactory = + new RowDataStoreWriteOperator.Factory( + table, + (fileStoreTable, + initialCommitUser, + state, + ioManager, + memoryPool, + metricGroup) -> + new StoreSinkWriteImpl( + fileStoreTable, + initialCommitUser, + state, + ioManager, + false, + false, + true, + memoryPool, + metricGroup), + commitUser) { + @Override + @SuppressWarnings("unchecked") + public > T createStreamOperator( + StreamOperatorParameters parameters) { + return (T) + new CoordinatorCommittingRowDataStoreWriteOperator( + parameters, + table, + storeSinkWriteProvider, + commitUser, + gateway); + } + + @Override + @SuppressWarnings("rawtypes") + public Class getStreamOperatorClass( + ClassLoader classLoader) { + return CoordinatorCommittingRowDataStoreWriteOperator.class; + } + }; + InternalTypeInfo typeInfo = + new InternalTypeInfo<>(new InternalRowTypeSerializer(table.rowType())); + return new OneInputStreamOperatorTestHarness<>( + operatorFactory, typeInfo.createSerializer(new ExecutionConfig())); + } + + private static class TestingContext implements OperatorCoordinator.Context { + + @Override + public OperatorID getOperatorId() { + return new OperatorID(); + } + + public JobID getJobID() { + return new JobID(); + } + + @Override + public org.apache.flink.metrics.groups.OperatorCoordinatorMetricGroup metricGroup() { + return null; + } + + @Override + public void failJob(Throwable cause) { + throw new FlinkRuntimeException(cause); + } + + @Override + public int currentParallelism() { + return 1; + } + + @Override + public ClassLoader getUserCodeClassloader() { + return Thread.currentThread().getContextClassLoader(); + } + + @Override + public CoordinatorStore getCoordinatorStore() { + return null; + } + + @Override + public boolean isConcurrentExecutionAttemptsSupported() { + return false; + } + + @Nullable + @Override + public CheckpointCoordinator getCheckpointCoordinator() { + return null; + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java new file mode 100644 index 000000000000..91110d996218 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java @@ -0,0 +1,161 @@ +/* + * 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; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.flink.FlinkConnectorOptions; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.apache.flink.streaming.api.environment.CheckpointConfig; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.function.Consumer; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link FlinkSink}. */ +public class FlinkSinkTest extends CommitterOperatorTestBase { + + private static final RowType ROW_TYPE = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.BIGINT()}, new String[] {"a", "b"}); + + @Test + public void testCoordinatorCommitPreconditionsHappyPath() throws Exception { + FileStoreTable table = createUnawareBucketTable(options -> {}); + FlinkSink.checkCoordinatorCommitPreconditions(table, newCheckpointConfig(1), true); + } + + @Test + public void testCoordinatorCommitPreconditionsRejectsBatchOrNoCheckpoint() throws Exception { + FileStoreTable table = createUnawareBucketTable(options -> {}); + assertThatThrownBy( + () -> + FlinkSink.checkCoordinatorCommitPreconditions( + table, newCheckpointConfig(1), false)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testCoordinatorCommitPreconditionsRejectsPrimaryKeyTable() throws Exception { + FileStoreTable table = createPrimaryKeyTable(); + assertThatThrownBy( + () -> + FlinkSink.checkCoordinatorCommitPreconditions( + table, newCheckpointConfig(1), true)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testCoordinatorCommitPreconditionsRejectsFixedBucket() throws Exception { + FileStoreTable table = createFileStoreTable(); + assertThatThrownBy( + () -> + FlinkSink.checkCoordinatorCommitPreconditions( + table, newCheckpointConfig(1), true)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testCoordinatorCommitPreconditionsRejectsNonWriteOnly() throws Exception { + FileStoreTable table = + createUnawareBucketTable(options -> options.set(CoreOptions.WRITE_ONLY, false)); + assertThatThrownBy( + () -> + FlinkSink.checkCoordinatorCommitPreconditions( + table, newCheckpointConfig(1), true)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testCoordinatorCommitPreconditionsRejectsPrecommitCompact() throws Exception { + FileStoreTable table = + createUnawareBucketTable( + options -> options.set(FlinkConnectorOptions.PRECOMMIT_COMPACT, true)); + assertThatThrownBy( + () -> + FlinkSink.checkCoordinatorCommitPreconditions( + table, newCheckpointConfig(1), true)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testCoordinatorCommitPreconditionsRejectsAutoTagForSavepoint() throws Exception { + FileStoreTable table = + createUnawareBucketTable( + options -> + options.set( + FlinkConnectorOptions.SINK_AUTO_TAG_FOR_SAVEPOINT, true)); + assertThatThrownBy( + () -> + FlinkSink.checkCoordinatorCommitPreconditions( + table, newCheckpointConfig(1), true)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testCoordinatorCommitPreconditionsRejectsConcurrentCheckpoints() throws Exception { + FileStoreTable table = createUnawareBucketTable(options -> {}); + assertThatThrownBy( + () -> + FlinkSink.checkCoordinatorCommitPreconditions( + table, newCheckpointConfig(2), true)) + .isInstanceOf(IllegalArgumentException.class); + } + + private FileStoreTable createUnawareBucketTable(Consumer setOptions) throws Exception { + return createFileStoreTable( + options -> { + options.set(CoreOptions.BUCKET, -1); + options.remove("bucket-key"); + options.set(CoreOptions.WRITE_ONLY, true); + setOptions.accept(options); + }); + } + + private FileStoreTable createPrimaryKeyTable() throws Exception { + Options conf = new Options(); + conf.set(CoreOptions.PATH, tablePath.toString()); + conf.setString("bucket", "1"); + SchemaManager schemaManager = new SchemaManager(LocalFileIO.create(), tablePath); + schemaManager.createTable( + new Schema( + ROW_TYPE.getFields(), + Collections.emptyList(), + Collections.singletonList("a"), + conf.toMap(), + "")); + return FileStoreTableFactory.create(LocalFileIO.create(), conf); + } + + private static CheckpointConfig newCheckpointConfig(int maxConcurrentCheckpoints) { + CheckpointConfig config = new CheckpointConfig(); + config.setMaxConcurrentCheckpoints(maxConcurrentCheckpoints); + return config; + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittableEventTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittableEventTest.java new file mode 100644 index 000000000000..d7a632671b9c --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittableEventTest.java @@ -0,0 +1,90 @@ +/* + * 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.data.BinaryRow; +import org.apache.paimon.data.BinaryRowWriter; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.flink.sink.Committable; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; + +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.List; + +import static org.apache.paimon.flink.sink.coordinator.CommittableEvent.createCommittableListSerializer; +import static org.apache.paimon.flink.sink.coordinator.WriterCommittablesTest.committableEquals; +import static org.apache.paimon.manifest.ManifestCommittableSerializerTest.randomCompactIncrement; +import static org.apache.paimon.manifest.ManifestCommittableSerializerTest.randomNewFilesIncrement; +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit test for {@link CommittableEvent}. */ +public class CommittableEventTest { + @Test + public void testSerialization() throws Exception { + ListSerializer serializer = createCommittableListSerializer(); + + DataIncrement dataIncrement = randomNewFilesIncrement(); + CompactIncrement compactIncrement = randomCompactIncrement(); + CommitMessage commitMessage = + new CommitMessageImpl(createTestRow(), 1, 2, dataIncrement, compactIncrement); + long checkpointId = 123L; + Committable committable = new Committable(checkpointId, commitMessage); + CommittableEvent event = + CommittableEvent.create( + checkpointId, true, Collections.singletonList(committable), serializer); + + assertThat(event.getCheckpointId()).isEqualTo(checkpointId); + assertThat(event.isRestoring()).isTrue(); + DataInputDeserializer in = new DataInputDeserializer(event.getSerialized()); + List resultCommittables = serializer.deserialize(in); + assertThat(resultCommittables.size()).isEqualTo(1); + assertThat(committableEquals(resultCommittables.get(0), committable)).isTrue(); + } + + @Test + public void testSerializationWithEmptyCommittable() throws Exception { + ListSerializer serializer = createCommittableListSerializer(); + + long checkpointId = 123L; + CommittableEvent event = + CommittableEvent.create(checkpointId, false, Collections.emptyList(), serializer); + + assertThat(event.getCheckpointId()).isEqualTo(checkpointId); + assertThat(event.isRestoring()).isFalse(); + DataInputDeserializer in = new DataInputDeserializer(event.getSerialized()); + List resultCommittables = serializer.deserialize(in); + assertThat(resultCommittables.isEmpty()).isTrue(); + } + + private static BinaryRow createTestRow() { + BinaryRow row = new BinaryRow(2); + BinaryRowWriter writer = new BinaryRowWriter(row); + writer.writeInt(0, 1024); + writer.writeString(1, BinaryString.fromString("abc")); + writer.complete(); + return row; + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java new file mode 100644 index 000000000000..83c4f9fbfbc9 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java @@ -0,0 +1,1094 @@ +/* + * 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.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryRowWriter; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.flink.FlinkConnectorOptions; +import org.apache.paimon.flink.sink.Committable; +import org.apache.paimon.flink.sink.Committer; +import org.apache.paimon.flink.sink.CommitterOperatorTestBase; +import org.apache.paimon.flink.sink.StoreCommitter; +import org.apache.paimon.flink.sink.state.CoordinatorState; +import org.apache.paimon.flink.sink.state.CoordinatorStateSerializer; +import org.apache.paimon.flink.sink.state.MemoryBackendStateStore; +import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.StreamTableCommit; +import org.apache.paimon.table.sink.StreamTableWrite; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.core.io.SimpleVersionedSerialization; +import org.apache.flink.metrics.groups.OperatorCoordinatorMetricGroup; +import org.apache.flink.runtime.checkpoint.CheckpointCoordinator; +import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.messages.Acknowledge; +import org.apache.flink.runtime.operators.coordination.CoordinatorStore; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Unit tests for {@link CommittingWriteOperatorCoordinator}. */ +public class CommittingWriteOperatorCoordinatorTest extends CommitterOperatorTestBase { + + private static final ListSerializer SERIALIZER = + CommittableEvent.createCommittableListSerializer(); + + private String commitUser; + private volatile Throwable failureCause; + + @BeforeEach + public void before() { + super.before(); + commitUser = UUID.randomUUID().toString(); + failureCause = null; + } + + @AfterEach + public void checkNoFailure() { + assertThat(failureCause).isNull(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testCommitSingleSubtask() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 1); + CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false); + coordinator.start(); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + + coordinator.handleEventFromOperator(0, 0, event(false, committable(table, 1, 1))); + coordinator.notifyCheckpointComplete(1); + coordinator.waitProcessAllActions(); + + assertResults(table, "1, 1"); + coordinator.close(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.CLOSED); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testCommitFanInFromMultipleSubtasks() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 2); + CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false); + coordinator.start(); + coordinator.waitProcessAllActions(); + + coordinator.handleEventFromOperator(0, 0, event(false, committable(table, 1, 1))); + coordinator.handleEventFromOperator(1, 0, event(false, committable(table, 1, 2))); + coordinator.notifyCheckpointComplete(1); + coordinator.waitProcessAllActions(); + + assertResults(table, "1, 1", "2, 2"); + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testWatermarkCommit() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 1); + CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false); + coordinator.start(); + coordinator.waitProcessAllActions(); + + coordinator.handleEventFromOperator(0, 0, event(false, committable(table, 1, 1))); + coordinator.handleEventFromOperator(0, 0, new WatermarkEvent(1024L)); + coordinator.notifyCheckpointComplete(1); + coordinator.waitProcessAllActions(); + assertThat(table.snapshotManager().latestSnapshot().watermark()).isEqualTo(1024L); + + // Long.MAX_VALUE watermark is ignored, so the committed watermark stays at 1024 + coordinator.handleEventFromOperator(0, 0, event(false, committable(table, 2, 2))); + coordinator.handleEventFromOperator(0, 0, new WatermarkEvent(Long.MAX_VALUE)); + coordinator.notifyCheckpointComplete(2); + coordinator.waitProcessAllActions(); + assertThat(table.snapshotManager().latestSnapshot().watermark()).isEqualTo(1024L); + + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testRestoringAlignsBeforeRunning() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 2); + + // first incarnation commits checkpoint 1 and captures the coordinator state + CommittingWriteOperatorCoordinator first = createCoordinator(table, context, false); + first.start(); + first.waitProcessAllActions(); + first.handleEventFromOperator(0, 0, event(false, committable(table, 1, 1))); + first.handleEventFromOperator(1, 0, event(false, committable(table, 1, 2))); + CompletableFuture checkpoint = new CompletableFuture<>(); + first.checkpointCoordinator(1, checkpoint); + first.notifyCheckpointComplete(1); + first.waitProcessAllActions(); + byte[] state = checkpoint.get(); + first.close(); + assertResults(table, "1, 1", "2, 2"); + + // second incarnation restores and stays RESTORING until both subtasks re-emit + CommittingWriteOperatorCoordinator second = createCoordinator(table, context, false); + second.resetToCheckpoint(1, state); + assertThat(second.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RESTORING); + second.start(); + second.waitProcessAllActions(); + assertThat(second.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RESTORING); + assertThat(second.getCommitUser()).isEqualTo(commitUser); + + second.handleEventFromOperator(0, 1, event(true, committable(table, 1, 3))); + second.waitProcessAllActions(); + assertThat(second.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RESTORING); + + second.handleEventFromOperator(1, 1, event(true, committable(table, 1, 4))); + second.waitProcessAllActions(); + assertThat(second.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + + // abandon path: restoring committables are dropped, not recommitted + assertResults(table, "1, 1", "2, 2"); + second.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testSnapshotLostWhenFailed() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 1); + + // first incarnation: cp1 fully committed, cp2 snapshotted but never notified + CommittingWriteOperatorCoordinator first = createCoordinator(table, context, false); + first.start(); + first.waitProcessAllActions(); + first.handleEventFromOperator(0, 0, event(false, committable(table, 1, 1))); + first.notifyCheckpointComplete(1L); + first.waitProcessAllActions(); + assertResults(table, "1, 1"); + + first.handleEventFromOperator(0, 0, event(false, committable(table, 2, 2))); + CompletableFuture cp2State = new CompletableFuture<>(); + first.checkpointCoordinator(2L, cp2State); + first.waitProcessAllActions(); + byte[] state = cp2State.get(); + first.close(); + // cp2 was never notified — only cp1 is in the table + assertResults(table, "1, 1"); + + // second incarnation: restore from cp2 state, replay the cp2 restoring event. abandon + // mode drops it; the snapshot from cp1 stays untouched. + CommittingWriteOperatorCoordinator second = createCoordinator(table, context, false); + second.resetToCheckpoint(2L, state); + second.start(); + second.waitProcessAllActions(); + second.handleEventFromOperator(0, 0, event(true, committable(table, 2, 2))); + second.waitProcessAllActions(); + assertThat(second.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + assertResults(table, "1, 1"); + + // a fresh checkpoint after recovery commits normally + second.handleEventFromOperator(0, 0, event(false, committable(table, 3, 3))); + second.notifyCheckpointComplete(3L); + second.waitProcessAllActions(); + assertResults(table, "1, 1", "3, 3"); + second.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testRejectCheckpointWhileRestoring() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 1); + CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false); + coordinator.resetToCheckpoint(2, emptyState()); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RESTORING); + coordinator.start(); + coordinator.waitProcessAllActions(); + + CompletableFuture checkpoint = new CompletableFuture<>(); + coordinator.checkpointCoordinator(3, checkpoint); + coordinator.waitProcessAllActions(); + assertThat(checkpoint.isCompletedExceptionally()).isTrue(); + + coordinator.handleEventFromOperator(0, 0, event(true, committable(table, 2, 1))); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testFailIntentionallyAfterRestoring() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 1); + + // capture coordinator state without committing checkpoint 1 + CommittingWriteOperatorCoordinator first = createCoordinator(table, context, true); + first.start(); + first.handleEventFromOperator(0, 0, event(false, committable(table, 1, 1))); + CompletableFuture checkpoint = new CompletableFuture<>(); + first.checkpointCoordinator(1, checkpoint); + first.waitProcessAllActions(); + byte[] state = checkpoint.get(); + first.close(); + // checkpoint 1 was never committed + assertThat(table.latestSnapshot()).isNotPresent(); + + // restore with failoverAfterRecovery: the restored committables are recommitted and an + // intentional failure is raised to reinitialize all writers + CommittingWriteOperatorCoordinator second = createCoordinator(table, context, true); + second.resetToCheckpoint(1, state); + second.start(); + second.waitProcessAllActions(); + second.handleEventFromOperator(0, 0, event(true, committable(table, 1, 1))); + second.waitProcessAllActions(); + + assertThat(failureCause).isInstanceOf(RuntimeException.class); + assertThat(failureCause).hasMessageContaining("intentionally thrown"); + assertResults(table, "1, 1"); + failureCause = null; + second.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testCheckpointAbort() throws Exception { + // accumulate committables across many checkpoints without notification, then notify only + // the last one. the coordinator must drain all pending checkpoints in a single commit. + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 1); + CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false); + coordinator.start(); + coordinator.waitProcessAllActions(); + + long lastCp = 10L; + List expected = new ArrayList<>(); + for (long cp = 1; cp <= lastCp; cp++) { + int value = (int) cp; + coordinator.handleEventFromOperator(0, 0, event(false, committable(table, cp, value))); + expected.add(value + ", " + value); + } + coordinator.notifyCheckpointComplete(lastCp); + coordinator.waitProcessAllActions(); + + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + assertThat(snapshot).isNotNull(); + assertThat(snapshot.commitIdentifier()).isEqualTo(lastCp); + Collections.sort(expected); + assertResults(table, expected.toArray(new String[0])); + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testCheckpointFutureCompletedExceptionallyOnSnapshotFailure() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 1); + RuntimeException expected = new RuntimeException("snapshotState boom"); + CommittingWriteOperatorCoordinator coordinator = + new CommittingWriteOperatorCoordinator( + context, + commitContext -> + new FailingSnapshotCommitter( + new StoreCommitter( + table, + table.newStreamWriteBuilder() + .withCommitUser(commitContext.commitUser()) + .newCommit(), + commitContext), + expected), + true, + commitUser, + false); + coordinator.start(); + coordinator.waitProcessAllActions(); + + CompletableFuture checkpoint = new CompletableFuture<>(); + coordinator.checkpointCoordinator(1L, checkpoint); + coordinator.waitProcessAllActions(); + + assertThat(checkpoint.isCompletedExceptionally()).isTrue(); + assertThatThrownBy(checkpoint::get).hasCause(expected); + assertThat(failureCause).isSameAs(expected); + failureCause = null; + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testEmptyCommit() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 1); + CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false); + coordinator.start(); + coordinator.waitProcessAllActions(); + + coordinator.handleEventFromOperator( + 0, 0, CommittableEvent.create(1L, false, Collections.emptyList(), SERIALIZER)); + coordinator.notifyCheckpointComplete(1L); + coordinator.waitProcessAllActions(); + + assertThat(table.snapshotManager().latestSnapshot()).isNull(); + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testForceCreateSnapshotCommit() throws Exception { + FileStoreTable table = + createFileStoreTable( + options -> { + options.set(CoreOptions.BUCKET, -1); + options.remove("bucket-key"); + options.set(CoreOptions.COMMIT_FORCE_CREATE_SNAPSHOT, true); + }); + TestingContext context = new TestingContext(new OperatorID(), 1); + CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false); + coordinator.start(); + coordinator.waitProcessAllActions(); + + coordinator.handleEventFromOperator( + 0, 0, CommittableEvent.create(1L, false, Collections.emptyList(), SERIALIZER)); + coordinator.notifyCheckpointComplete(1L); + coordinator.waitProcessAllActions(); + + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + assertThat(snapshot).isNotNull(); + assertThat(snapshot.commitIdentifier()).isEqualTo(1L); + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testPollManifestCommittablesForCheckpoint() throws Exception { + String partitionKey = "a"; + int partitionValue = 100; + long normalValue = 0L; + FileStoreTable table = + createFileStoreTable( + options -> { + options.set(CoreOptions.BUCKET, -1); + options.remove("bucket-key"); + }, + Collections.singletonList(partitionKey)); + long checkpointId1 = 1024L; + long checkpointId2 = 1025L; + long checkpointId3 = 1027L; + long watermark = System.currentTimeMillis(); + WriterCommittables[] writerCommittables = new WriterCommittables[3]; + // cp1, cp3 in subtask-0 + writerCommittables[0] = + new WriterCommittables( + checkpointId1, + Collections.singletonList( + committable(table, checkpointId1, partitionValue, normalValue++))); + writerCommittables[0].mergeWith( + new WriterCommittables(checkpointId2, Collections.emptyList())); + writerCommittables[0].mergeWith( + new WriterCommittables( + checkpointId3, + Collections.singletonList( + committable(table, checkpointId3, partitionValue, normalValue++)))); + // cp1, cp2 in subtask-1 + writerCommittables[1] = + new WriterCommittables( + checkpointId1, + Collections.singletonList( + committable(table, checkpointId1, partitionValue, normalValue++))); + writerCommittables[1].mergeWith( + new WriterCommittables( + checkpointId2, + Collections.singletonList( + committable(table, checkpointId2, partitionValue, normalValue++)))); + writerCommittables[1].mergeWith( + new WriterCommittables(checkpointId3, Collections.emptyList())); + // cp2, cp3 in subtask-2 + writerCommittables[2] = new WriterCommittables(checkpointId1, Collections.emptyList()); + writerCommittables[2].mergeWith( + new WriterCommittables( + checkpointId2, + Collections.singletonList( + committable(table, checkpointId2, partitionValue, normalValue++)))); + writerCommittables[2].mergeWith( + new WriterCommittables( + checkpointId3, + Collections.singletonList( + committable(table, checkpointId3, partitionValue, normalValue++)))); + + StreamTableCommit commit = table.newCommit(commitUser); + StoreCommitter committer = + new StoreCommitter( + table, commit, Committer.createContext("", null, true, false, null, 1, 0)); + + NavigableMap result = + CommittingWriteOperatorCoordinator.pollManifestCommittablesForCheckpoint( + checkpointId2, writerCommittables, committer, watermark); + + BinaryRow partition = new BinaryRow(1); + BinaryRowWriter writer = new BinaryRowWriter(partition); + writer.writeInt(0, partitionValue); + writer.complete(); + // verify result + assertThat(result.size()).isEqualTo(2); + // verify increasing order + List manifestCommittables = new ArrayList<>(result.values()); + assertThat(manifestCommittables.get(0).identifier()).isEqualTo(checkpointId1); + assertThat(manifestCommittables.get(1).identifier()).isEqualTo(checkpointId2); + + assertThat(result.get(checkpointId1)).isNotNull(); + assertThat(result.get(checkpointId1).identifier()).isEqualTo(checkpointId1); + assertThat(result.get(checkpointId1).watermark()).isEqualTo(watermark); + assertThat(result.get(checkpointId1).fileCommittables().size()).isEqualTo(2); + assertThat(result.get(checkpointId1).fileCommittables().get(0).partition()) + .isEqualTo(partition); + assertThat(result.get(checkpointId1).fileCommittables().get(1).partition()) + .isEqualTo(partition); + + assertThat(result.get(checkpointId2)).isNotNull(); + assertThat(result.get(checkpointId2).identifier()).isEqualTo(checkpointId2); + assertThat(result.get(checkpointId2).watermark()).isEqualTo(watermark); + assertThat(result.get(checkpointId2).fileCommittables().size()).isEqualTo(2); + assertThat(result.get(checkpointId2).fileCommittables().get(0).partition()) + .isEqualTo(partition); + assertThat(result.get(checkpointId2).fileCommittables().get(1).partition()) + .isEqualTo(partition); + + assertThat(result.get(checkpointId3)).isNull(); + + // verify remaining subtask committables + assertThat(writerCommittables[0].getCommittablesPerCheckpoint().size()).isEqualTo(1); + assertThat(writerCommittables[0].getCommittablesPerCheckpoint().get(checkpointId3)) + .isNotNull(); + assertThat(writerCommittables[0].getCommittablesPerCheckpoint().get(checkpointId3).size()) + .isEqualTo(1); + assertThat(writerCommittables[1].getCommittablesPerCheckpoint().isEmpty()).isTrue(); + assertThat(writerCommittables[2].getCommittablesPerCheckpoint().size()).isEqualTo(1); + assertThat(writerCommittables[2].getCommittablesPerCheckpoint().get(checkpointId3)) + .isNotNull(); + assertThat(writerCommittables[2].getCommittablesPerCheckpoint().get(checkpointId3).size()) + .isEqualTo(1); + + committer.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testPartialFailoverWithoutRestoring() throws Exception { + // non-restore -> trigger checkpoint -> partial failover -> checkpoint abort -> re-trigger + // checkpoint -> checkpoint complete + FileStoreTable table = + createFileStoreTable( + options -> { + options.set(CoreOptions.BUCKET, -1); + options.remove("bucket-key"); + options.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT.key(), "0"); + }, + Collections.singletonList("a")); + TestingContext context = new TestingContext(new OperatorID(), 2); + CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false); + // 1. start with non-restoring + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.CREATED); + coordinator.start(); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + // 2. trigger checkpoint 1 + long checkpointId = 1L; + // checkpoint coordinator before task + { + CompletableFuture checkpointFuture = new CompletableFuture<>(); + coordinator.checkpointCoordinator(checkpointId, checkpointFuture); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + assertThat(checkpointFuture.isDone()).isTrue(); + assertThat(checkpointFuture.isCompletedExceptionally()).isFalse(); + } + // write data and checkpoint task-0 + coordinator.handleEventFromOperator( + 0, + 0, + eventOf( + checkpointId, + false, + committables( + table, checkpointId, GenericRow.of(1, 2L), GenericRow.of(1, 3L)))); + // write data and checkpoint task-1 + coordinator.handleEventFromOperator( + 1, + 0, + eventOf( + checkpointId, + false, + committables( + table, checkpointId, GenericRow.of(1, 4L), GenericRow.of(1, 5L)))); + // 3. partial failover, fail task-0 + coordinator.executionAttemptFailed(0, 0, new Exception("Fail subtask 0 as expected")); + coordinator.executionAttemptReady(0, 1, new MockSubtaskGateway()); + coordinator.subtaskReset(0, -1); + // 4. checkpoint 1 abort + coordinator.notifyCheckpointAborted(checkpointId); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + + // 5. re-trigger checkpoint + checkpointId++; + { + CompletableFuture checkpointFuture = new CompletableFuture<>(); + coordinator.checkpointCoordinator(checkpointId, checkpointFuture); + coordinator.waitProcessAllActions(); + assertThat(checkpointFuture.isDone()).isTrue(); + assertThat(checkpointFuture.isCompletedExceptionally()).isFalse(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + } + // rewrite some data and checkpoint task-0 + coordinator.handleEventFromOperator( + 0, + 1, + eventOf( + checkpointId, + false, + committables( + table, + checkpointId, + GenericRow.of(1, 2L), + GenericRow.of(1, 3L), + GenericRow.of(1, 6L)))); + // write empty data and checkpoint task-1 + coordinator.handleEventFromOperator( + 1, + 0, + CommittableEvent.create(checkpointId, false, Collections.emptyList(), SERIALIZER)); + // notify cp complete + coordinator.notifyCheckpointComplete(checkpointId); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + assertResults(table, "1, 2", "1, 3", "1, 4", "1, 5", "1, 6"); + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testPartialFailoverWithRestoring() throws Exception { + // non-restore -> trigger checkpoint -> checkpoint complete -> partial failover -> + // restore subtask -> trigger checkpoint -> checkpoint complete + FileStoreTable table = + createFileStoreTable( + options -> { + options.set(CoreOptions.BUCKET, -1); + options.remove("bucket-key"); + // set manifest merge min count to 0 to find repeated immediately + options.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT.key(), "0"); + }, + Collections.singletonList("a")); + TestingContext context = new TestingContext(new OperatorID(), 2); + CommittingWriteOperatorCoordinator coordinator = createCoordinator(table, context, false); + // 1. start with non-restoring + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.CREATED); + coordinator.start(); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + // 2. trigger checkpoint 1 + long checkpointId = 1L; + // checkpoint coordinator before task + { + CompletableFuture checkpointFuture = new CompletableFuture<>(); + coordinator.checkpointCoordinator(checkpointId, checkpointFuture); + coordinator.waitProcessAllActions(); + assertThat(checkpointFuture.isDone()).isTrue(); + assertThat(checkpointFuture.isCompletedExceptionally()).isFalse(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + } + // write data and checkpoint task-0 + List subtask0Committables = + committables(table, checkpointId, GenericRow.of(1, 2L), GenericRow.of(1, 3L)); + coordinator.handleEventFromOperator( + 0, 0, eventOf(checkpointId, false, subtask0Committables)); + CommittableEvent restoreEvent = + CommittableEvent.create(checkpointId, true, subtask0Committables, SERIALIZER); + // write data and checkpoint task-1 + coordinator.handleEventFromOperator( + 1, + 0, + eventOf( + checkpointId, + false, + committables( + table, checkpointId, GenericRow.of(1, 4L), GenericRow.of(1, 5L)))); + // 3. checkpoint complete + coordinator.notifyCheckpointComplete(checkpointId); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + // 4. partial failover, fail task-0 + coordinator.executionAttemptFailed(0, 0, new Exception("Fail subtask 0 as expected")); + coordinator.executionAttemptReady(0, 1, new MockSubtaskGateway()); + coordinator.subtaskReset(0, checkpointId); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + + // 5. restore subtask-0 + coordinator.handleEventFromOperator(0, 1, restoreEvent); + + // 6. re-trigger checkpoint + checkpointId++; + { + CompletableFuture checkpointFuture = new CompletableFuture<>(); + coordinator.checkpointCoordinator(checkpointId, checkpointFuture); + coordinator.waitProcessAllActions(); + assertThat(checkpointFuture.isDone()).isTrue(); + assertThat(checkpointFuture.isCompletedExceptionally()).isFalse(); + } + // rewrite some data and checkpoint task-0 + coordinator.handleEventFromOperator( + 0, + 1, + eventOf( + checkpointId, + false, + committables( + table, + checkpointId, + GenericRow.of(1, 6L), + GenericRow.of(1, 7L), + GenericRow.of(1, 8L)))); + // write empty data and checkpoint task-1 + coordinator.handleEventFromOperator( + 1, + 0, + CommittableEvent.create(checkpointId, false, Collections.emptyList(), SERIALIZER)); + // 7. notify cp complete + coordinator.notifyCheckpointComplete(checkpointId); + coordinator.waitProcessAllActions(); + assertThat(coordinator.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + assertResults(table, "1, 2", "1, 3", "1, 4", "1, 5", "1, 6", "1, 7", "1, 8"); + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testRestoreEmptyMarkDoneState() throws Exception { + // mark-done introduces a partition listener that is initialized lazily from coordinator + // state. restoring an empty mark-done state must not crash the coordinator. + Map markDoneOption = new HashMap<>(); + markDoneOption.put(FlinkConnectorOptions.PARTITION_IDLE_TIME_TO_DONE.key(), "1h"); + + FileStoreTable table = + createFileStoreTable( + options -> { + options.set(CoreOptions.BUCKET, -1); + options.remove("bucket-key"); + }, + Collections.singletonList("a")); + TestingContext context = new TestingContext(new OperatorID(), 1); + + // 1. capture state from a coordinator without mark-done enabled + CommittingWriteOperatorCoordinator first = createCoordinator(table, context, false); + first.start(); + first.waitProcessAllActions(); + first.handleEventFromOperator(0, 0, event(false, committable(table, 1, 1))); + CompletableFuture checkpoint = new CompletableFuture<>(); + first.checkpointCoordinator(1L, checkpoint); + first.notifyCheckpointComplete(1L); + first.waitProcessAllActions(); + byte[] state = checkpoint.get(); + first.close(); + + // 2. restore with mark-done enabled — should initialize cleanly + FileStoreTable markDoneTable = table.copy(markDoneOption); + CommittingWriteOperatorCoordinator second = + createCoordinator(markDoneTable, context, false); + second.resetToCheckpoint(1L, state); + second.start(); + second.waitProcessAllActions(); + second.handleEventFromOperator(0, 1, event(true, committable(markDoneTable, 1, 1))); + second.waitProcessAllActions(); + assertThat(second.getCurrentState()) + .isEqualTo(CommittingWriteOperatorCoordinator.State.RUNNING); + second.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testWriteRestoreOnlyCoordinatorIgnoresCommit() throws Exception { + // the base WriteOperatorCoordinator answers checkpoint with empty bytes and ignores + // committable events, preserving the pure write-restore behavior. + FileStoreTable table = createUnawareBucketTable(); + WriteOperatorCoordinator coordinator = new WriteOperatorCoordinator(table); + coordinator.start(); + CompletableFuture checkpoint = new CompletableFuture<>(); + coordinator.checkpointCoordinator(1, checkpoint); + assertThat(checkpoint.get()).isEmpty(); + coordinator.close(); + } + + /** + * The coordinator runs at parallelism 1 (single instance per JobVertex), so the {@link + * Committer.Context} it hands to the committer must always report parallelism=1 and + * subtaskIndex=0, regardless of the writer operator's parallelism. + */ + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testCommitterContextParallelism() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + // deliberately use a writer parallelism > 1 to catch the "writer parallelism leaks into + // Committer.Context" bug pattern. + TestingContext context = new TestingContext(new OperatorID(), 8); + AtomicReference captured = new AtomicReference<>(); + CommittingWriteOperatorCoordinator coordinator = + createCoordinatorCapturingContext(table, context, captured); + coordinator.start(); + coordinator.waitProcessAllActions(); + + Committer.Context committerContext = captured.get(); + assertThat(committerContext).isNotNull(); + assertThat(committerContext.getParallelism()).isEqualTo(1); + assertThat(committerContext.getSubtaskIndex()).isEqualTo(0); + coordinator.close(); + } + + @Timeout(value = 30, unit = TimeUnit.SECONDS) + @Test + public void testCommitterContextIsRestored() throws Exception { + FileStoreTable table = createUnawareBucketTable(); + TestingContext context = new TestingContext(new OperatorID(), 2); + + // fresh start: no prior coordinator state, isRestored must be false + AtomicReference freshContext = new AtomicReference<>(); + CommittingWriteOperatorCoordinator fresh = + createCoordinatorCapturingContext(table, context, freshContext); + fresh.start(); + fresh.waitProcessAllActions(); + assertThat(freshContext.get()).isNotNull(); + assertThat(freshContext.get().isRestored()).isFalse(); + // produce a committed checkpoint so we have real state bytes to restore from + fresh.handleEventFromOperator(0, 0, event(false, committable(table, 1, 1))); + fresh.handleEventFromOperator(1, 0, event(false, committable(table, 1, 2))); + CompletableFuture checkpoint = new CompletableFuture<>(); + fresh.checkpointCoordinator(1, checkpoint); + fresh.notifyCheckpointComplete(1); + fresh.waitProcessAllActions(); + byte[] state = checkpoint.get(); + fresh.close(); + + // restored start: coordinator resets to checkpoint before start(), isRestored must be true + AtomicReference restoredContext = new AtomicReference<>(); + CommittingWriteOperatorCoordinator restored = + createCoordinatorCapturingContext(table, context, restoredContext); + restored.resetToCheckpoint(1, state); + restored.start(); + restored.waitProcessAllActions(); + assertThat(restoredContext.get()).isNotNull(); + assertThat(restoredContext.get().isRestored()).isTrue(); + restored.close(); + } + + // ------------------------------------------------------------------------ + + private FileStoreTable createUnawareBucketTable() throws Exception { + return createFileStoreTable( + options -> { + options.set(CoreOptions.BUCKET, -1); + options.remove("bucket-key"); + }); + } + + private CommittingWriteOperatorCoordinator createCoordinator( + FileStoreTable table, TestingContext context, boolean failoverAfterRecovery) { + return new CommittingWriteOperatorCoordinator( + context, + commitContext -> + new StoreCommitter( + table, + table.newStreamWriteBuilder() + .withCommitUser(commitContext.commitUser()) + .newCommit(), + commitContext), + true, + commitUser, + failoverAfterRecovery); + } + + private CommittingWriteOperatorCoordinator createCoordinatorCapturingContext( + FileStoreTable table, + TestingContext context, + AtomicReference captured) { + return new CommittingWriteOperatorCoordinator( + context, + commitContext -> { + captured.set(commitContext); + return new StoreCommitter( + table, + table.newStreamWriteBuilder() + .withCommitUser(commitContext.commitUser()) + .newCommit(), + commitContext); + }, + true, + commitUser, + false); + } + + private Committable committable(FileStoreTable table, long checkpointId, int value) + throws Exception { + try (StreamTableWrite write = + table.newStreamWriteBuilder().withCommitUser(commitUser).newWrite()) { + write.write(GenericRow.of(value, (long) value)); + List messages = write.prepareCommit(false, checkpointId); + assertThat(messages).hasSize(1); + return new Committable(checkpointId, messages.get(0)); + } + } + + private Committable committable( + FileStoreTable table, long checkpointId, int partitionValue, long normalValue) + throws Exception { + try (StreamTableWrite write = + table.newStreamWriteBuilder().withCommitUser(commitUser).newWrite()) { + write.write(GenericRow.of(partitionValue, normalValue)); + List messages = write.prepareCommit(false, checkpointId); + assertThat(messages).hasSize(1); + return new Committable(checkpointId, messages.get(0)); + } + } + + private List committables( + FileStoreTable table, long checkpointId, GenericRow... rows) throws Exception { + List result = new ArrayList<>(); + try (StreamTableWrite write = + table.newStreamWriteBuilder().withCommitUser(commitUser).newWrite()) { + for (GenericRow row : rows) { + write.write(row); + } + for (CommitMessage message : write.prepareCommit(false, checkpointId)) { + result.add(new Committable(checkpointId, message)); + } + } + return result; + } + + private CommittableEvent event(boolean isRestoring, Committable committable) throws Exception { + List list = new ArrayList<>(); + list.add(committable); + return CommittableEvent.create(committable.checkpointId(), isRestoring, list, SERIALIZER); + } + + private CommittableEvent eventOf( + long checkpointId, boolean isRestoring, List committables) + throws Exception { + return CommittableEvent.create(checkpointId, isRestoring, committables, SERIALIZER); + } + + private byte[] emptyState() throws Exception { + return SimpleVersionedSerialization.writeVersionAndSerialize( + new CoordinatorStateSerializer(), + new CoordinatorState( + commitUser, + Long.MIN_VALUE, + new MemoryBackendStateStore().getSerializedStates())); + } + + private class TestingContext implements OperatorCoordinator.Context { + + private final OperatorID operatorID; + private final int parallelism; + + private TestingContext(OperatorID operatorID, int parallelism) { + this.operatorID = operatorID; + this.parallelism = parallelism; + } + + @Override + public OperatorID getOperatorId() { + return operatorID; + } + + public JobID getJobID() { + return new JobID(); + } + + @Override + public OperatorCoordinatorMetricGroup metricGroup() { + return null; + } + + @Override + public void failJob(Throwable cause) { + failureCause = cause; + } + + @Override + public int currentParallelism() { + return parallelism; + } + + @Override + public ClassLoader getUserCodeClassloader() { + return Thread.currentThread().getContextClassLoader(); + } + + @Override + public CoordinatorStore getCoordinatorStore() { + return null; + } + + @Override + public boolean isConcurrentExecutionAttemptsSupported() { + return false; + } + + @Nullable + @Override + public CheckpointCoordinator getCheckpointCoordinator() { + return null; + } + } + + /** {@link Committer} decorator whose {@link #snapshotState()} always throws. */ + private static class FailingSnapshotCommitter + implements Committer { + + private final Committer delegate; + private final RuntimeException failure; + + FailingSnapshotCommitter( + Committer delegate, RuntimeException failure) { + this.delegate = delegate; + this.failure = failure; + } + + @Override + public void snapshotState() { + throw failure; + } + + @Override + public boolean forceCreatingSnapshot() { + return delegate.forceCreatingSnapshot(); + } + + @Override + public ManifestCommittable combine( + long checkpointId, long watermark, List committables) + throws IOException { + return delegate.combine(checkpointId, watermark, committables); + } + + @Override + public ManifestCommittable combine( + long checkpointId, + long watermark, + ManifestCommittable t, + List committables) { + return delegate.combine(checkpointId, watermark, t, committables); + } + + @Override + public void commit(List globalCommittables) + throws IOException, InterruptedException { + delegate.commit(globalCommittables); + } + + @Override + public int filterAndCommit( + List globalCommittables, + boolean checkAppendFiles, + boolean partitionMarkDoneRecoverFromState) + throws IOException { + return delegate.filterAndCommit( + globalCommittables, checkAppendFiles, partitionMarkDoneRecoverFromState); + } + + @Override + public Map> groupByCheckpoint( + Collection committables) { + return delegate.groupByCheckpoint(committables); + } + + @Override + public void close() throws Exception { + delegate.close(); + } + } + + private static class MockSubtaskGateway implements OperatorCoordinator.SubtaskGateway { + + @Override + public CompletableFuture sendEvent(OperatorEvent evt) { + throw new UnsupportedOperationException("Unsupported to send event " + evt); + } + + @Override + public ExecutionAttemptID getExecution() { + throw new UnsupportedOperationException("Unsupported to get execution"); + } + + @Override + public int getSubtask() { + throw new UnsupportedOperationException("Unsupported to get subtask"); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SimpleWatermarkValveTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SimpleWatermarkValveTest.java new file mode 100644 index 000000000000..8249ad675384 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/SimpleWatermarkValveTest.java @@ -0,0 +1,60 @@ +/* + * 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.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit test case of {@link SimpleWatermarkValve}. */ +public class SimpleWatermarkValveTest { + @Test + public void test() { + SimpleWatermarkValve watermarkValve = new SimpleWatermarkValve(3); + assertThat(watermarkValve.getCurrentWatermark()).isEqualTo(Long.MIN_VALUE); + // update task-0 + watermarkValve.updateSubtaskWatermark(0, 1L); + assertThat(watermarkValve.getCurrentWatermark()).isEqualTo(Long.MIN_VALUE); + // update task-1 + watermarkValve.updateSubtaskWatermark(1, 2L); + assertThat(watermarkValve.getCurrentWatermark()).isEqualTo(Long.MIN_VALUE); + // re-update task-0 + watermarkValve.updateSubtaskWatermark(0, 3L); + assertThat(watermarkValve.getCurrentWatermark()).isEqualTo(Long.MIN_VALUE); + // update task-2, it's aligned + watermarkValve.updateSubtaskWatermark(2, 4L); + assertThat(watermarkValve.getCurrentWatermark()).isEqualTo(2L); + // reset + watermarkValve.reset(0L); + assertThat(watermarkValve.getCurrentWatermark()).isEqualTo(0L); + // re-update all watermarks + watermarkValve.updateSubtaskWatermark(0, 101L); + watermarkValve.updateSubtaskWatermark(1, 100L); + watermarkValve.updateSubtaskWatermark(2, 102L); + assertThat(watermarkValve.getCurrentWatermark()).isEqualTo(100L); + // older watermark would be ignored + watermarkValve.updateSubtaskWatermark(1, 1L); + assertThat(watermarkValve.getCurrentWatermark()).isEqualTo(100L); + // Long.MAX_VALUE would be ignored + watermarkValve.updateSubtaskWatermark(0, Long.MAX_VALUE); + watermarkValve.updateSubtaskWatermark(1, Long.MAX_VALUE); + watermarkValve.updateSubtaskWatermark(2, Long.MAX_VALUE); + assertThat(watermarkValve.getCurrentWatermark()).isEqualTo(100L); + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WatermarkEventTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WatermarkEventTest.java new file mode 100644 index 000000000000..28ef5a590dff --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WatermarkEventTest.java @@ -0,0 +1,35 @@ +/* + * 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.flink.streaming.api.watermark.Watermark; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit test case of {@link WatermarkEvent}. */ +public class WatermarkEventTest { + @Test + public void test() { + long watermark = System.currentTimeMillis(); + WatermarkEvent event = new WatermarkEvent(new Watermark(watermark)); + assertThat(event.getWatermark()).isEqualTo(watermark); + assertThat(event).isEqualTo(new WatermarkEvent(watermark)); + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WriterCommittablesTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WriterCommittablesTest.java new file mode 100644 index 000000000000..31bde431fe52 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/WriterCommittablesTest.java @@ -0,0 +1,311 @@ +/* + * 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.data.BinaryRow; +import org.apache.paimon.flink.sink.Committable; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; + +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.NavigableMap; +import java.util.Objects; +import java.util.TreeMap; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** Unit tests for {@link WriterCommittables}. */ +public class WriterCommittablesTest { + + private static final ListSerializer serializer = + CommittableEvent.createCommittableListSerializer(); + + @Test + public void testGetCommittablesPerCheckpoint() throws Exception { + CommitMessage commitMessage = createEmptyCommitMessage(); + long checkpointId = 1L; + Committable committable = new Committable(checkpointId, commitMessage); + CommittableEvent event = + CommittableEvent.create( + checkpointId, false, Collections.singletonList(committable), serializer); + WriterCommittables committables = WriterCommittables.from(event, serializer); + NavigableMap> resultCommittables = + committables.getCommittablesPerCheckpoint(); + + assertThat(resultCommittables.size()).isEqualTo(1); + assertThat(resultCommittables.get(checkpointId).size()).isEqualTo(1); + + Committable resultCommittable = resultCommittables.get(checkpointId).get(0); + assertThat(resultCommittable.checkpointId()).isEqualTo(checkpointId); + assertThat(resultCommittable.commitMessage()).isEqualTo(commitMessage); + } + + @Test + public void testGetCommittablesBeforeCheckpoint() throws Exception { + CommitMessage commitMessage = createEmptyCommitMessage(); + long checkpointId = 4L; + Committable committable1 = new Committable(1L, commitMessage); + Committable committable2 = new Committable(2L, commitMessage); + Committable committable3 = new Committable(3L, commitMessage); + CommittableEvent event = + CommittableEvent.create( + checkpointId, + false, + Arrays.asList(committable1, committable3, committable2), + serializer); + WriterCommittables committables = WriterCommittables.from(event, serializer); + assertThat(committables.getCommittablesBeforeCheckpoint(-1, true).isEmpty()).isTrue(); + assertThat(committables.getCommittablesBeforeCheckpoint(1, false).isEmpty()).isTrue(); + assertThat(committables.getCommittablesBeforeCheckpoint(1, true).isEmpty()).isFalse(); + assertThat(committables.getCommittablesBeforeCheckpoint(3, false).size()).isEqualTo(2); + assertThat(committables.getCommittablesBeforeCheckpoint(4, false).size()).isEqualTo(3); + } + + @Test + public void testReset() throws Exception { + CommitMessage commitMessage = createEmptyCommitMessage(); + long checkpointId = 4L; + Committable committable1 = new Committable(1L, commitMessage); + Committable committable2 = new Committable(2L, commitMessage); + Committable committable3 = new Committable(3L, commitMessage); + CommittableEvent event = + CommittableEvent.create( + checkpointId, + true, + Arrays.asList(committable1, committable3, committable2), + serializer); + WriterCommittables committables = WriterCommittables.from(event, serializer); + committables.reset(); + assertThat(committables.getMaxCheckpointId()).isLessThan(0); + assertThat(committables.getCommittablesPerCheckpoint().isEmpty()).isTrue(); + } + + @Test + public void testGroupByCheckpoint() { + TreeMap> committablesPerCheckpoint = new TreeMap<>(); + List committables = new ArrayList<>(); + CommitMessage commitMessage = createEmptyCommitMessage(); + committables.add(new Committable(1L, commitMessage)); + committables.add(new Committable(2L, commitMessage)); + committables.add(new Committable(3L, commitMessage)); + committables.add(new Committable(1L, commitMessage)); + WriterCommittables.groupByCheckpoint(committablesPerCheckpoint, committables); + assertThat(committablesPerCheckpoint.size()).isEqualTo(3); + assertThat(committablesPerCheckpoint.get(1L).size()).isEqualTo(2); + assertThat(committablesPerCheckpoint.get(2L).size()).isEqualTo(1); + assertThat(committablesPerCheckpoint.get(3L).size()).isEqualTo(1); + } + + @Test + public void testMergeWith() throws Exception { + CommitMessage commitMessage = createEmptyCommitMessage(); + long checkpointId = 1L; + WriterCommittables committables; + // create original committables; + Committable committable = new Committable(checkpointId, commitMessage); + committables = + WriterCommittables.from( + CommittableEvent.create( + checkpointId, + true, + Collections.singletonList(committable), + serializer), + serializer); + // merge new checkpoint committables; + checkpointId++; + committables.mergeWith( + WriterCommittables.from( + CommittableEvent.create( + checkpointId, false, Collections.emptyList(), serializer), + serializer)); + // verify + NavigableMap> results = committables.getCommittablesPerCheckpoint(); + // checkpoint 2 is empty, it's ignored + assertThat(results.size()).isEqualTo(1); + assertThat(results.get(1L)).isNotNull(); + assertThat(results.get(1L).size()).isEqualTo(1); + assertThat(results.get(2L)).isNull(); + assertThat(committables.getMaxCheckpointId()).isEqualTo(2L); + // merge another new committables + checkpointId++; + Committable committable1 = new Committable(checkpointId, commitMessage); + Committable committable2 = new Committable(checkpointId, commitMessage); + committables.mergeWith( + WriterCommittables.from( + CommittableEvent.create( + checkpointId, + false, + Arrays.asList(committable1, committable2), + serializer), + serializer)); + // verify + assertThat(committables.getMaxCheckpointId()).isEqualTo(3L); + assertThat(committables.getCommittablesPerCheckpoint().size()).isEqualTo(2); + // verify checkpoint 1 + assertThat(committables.getCommittablesPerCheckpoint().get(1L)).isNotNull(); + assertThat(committables.getCommittablesPerCheckpoint().get(1L).size()).isEqualTo(1); + assertThat( + committableEquals( + committables.getCommittablesPerCheckpoint().get(1L).get(0), + committable)) + .isTrue(); + // verify checkpoint 2 + assertThat(committables.getCommittablesPerCheckpoint().get(2L)).isNull(); + // verify checkpoint 3 + assertThat(committables.getCommittablesPerCheckpoint().get(3L)).isNotNull(); + assertThat(committables.getCommittablesPerCheckpoint().get(3L).size()).isEqualTo(2); + assertThat( + committableEquals( + committables.getCommittablesPerCheckpoint().get(3L).get(0), + committable1)) + .isTrue(); + assertThat( + committableEquals( + committables.getCommittablesPerCheckpoint().get(3L).get(1), + committable2)) + .isTrue(); + } + + @Test + public void testInvalidMerge() throws Exception { + CommitMessage commitMessage = createEmptyCommitMessage(); + long checkpointId = 1024L; + WriterCommittables committables = + WriterCommittables.from( + CommittableEvent.create( + checkpointId, + true, + Collections.singletonList( + new Committable(checkpointId, commitMessage)), + serializer), + serializer); + // could not merge same checkpoint id + assertThrows(IllegalStateException.class, () -> committables.mergeWith(committables)); + + long oldCheckpointId = checkpointId - 1; + WriterCommittables oldCommittables = + WriterCommittables.from( + CommittableEvent.create( + oldCheckpointId, + false, + Collections.singletonList( + new Committable(oldCheckpointId, commitMessage)), + serializer), + serializer); + assertThrows(IllegalStateException.class, () -> committables.mergeWith(oldCommittables)); + } + + @Test + public void testClearCommittablesBeforeCheckpoint() throws Exception { + CommitMessage commitMessage = createEmptyCommitMessage(); + Committable committableCp1 = new Committable(1L, commitMessage); + Committable committableCp2 = new Committable(2L, commitMessage); + Committable committableCp3 = new Committable(3L, commitMessage); + CommittableEvent event = + CommittableEvent.create( + 3L, + true, + Arrays.asList(committableCp3, committableCp1, committableCp2), + serializer); + WriterCommittables committables = WriterCommittables.from(event, serializer); + // clear checkpoint < 1, nothing happens + committables.clearCommittablesBeforeCheckpoint(1L, false); + assertThat(committables.getMaxCheckpointId()).isEqualTo(3L); + assertThat(committables.getCommittablesPerCheckpoint().size()).isEqualTo(3); + // clear checkpoint <= 1, checkpoint 2 and 3 left + committables.clearCommittablesBeforeCheckpoint(1L, true); + assertThat(committables.getMaxCheckpointId()).isEqualTo(3L); + assertThat(committables.getCommittablesPerCheckpoint().size()).isEqualTo(2); + assertThat(committables.getCommittablesPerCheckpoint().get(1L)).isNull(); + assertThat(committables.getCommittablesPerCheckpoint().get(2L)).isNotNull(); + assertThat(committables.getCommittablesPerCheckpoint().get(2L).size()).isEqualTo(1); + assertThat( + committableEquals( + committables.getCommittablesPerCheckpoint().get(2L).get(0), + committableCp2)) + .isTrue(); + assertThat(committables.getCommittablesPerCheckpoint().get(3L)).isNotNull(); + assertThat(committables.getCommittablesPerCheckpoint().get(3L).size()).isEqualTo(1); + assertThat( + committableEquals( + committables.getCommittablesPerCheckpoint().get(3L).get(0), + committableCp3)) + .isTrue(); + // clear all committables + committables.clearCommittablesBeforeCheckpoint(10L, true); + assertThat(committables.getMaxCheckpointId()).isEqualTo(-1); + assertThat(committables.getCommittablesPerCheckpoint().isEmpty()).isTrue(); + } + + @Test + public void testInvalidInputCommittables() { + CommitMessage commitMessage = createEmptyCommitMessage(); + assertThrows( + IllegalStateException.class, + () -> + WriterCommittables.from( + CommittableEvent.create( + 1L, + false, + Collections.singletonList( + // invalid checkpoint id, it's bigger than max + // checkpoint params + new Committable(2L, commitMessage)), + serializer), + serializer)); + } + + private static CommitMessage createEmptyCommitMessage() { + return new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + 0, + 1, + new DataIncrement( + Collections.emptyList(), Collections.emptyList(), Collections.emptyList()), + new CompactIncrement( + Collections.emptyList(), Collections.emptyList(), Collections.emptyList())); + } + + // Committable does not implement 'equals' + public static boolean committableEquals(Committable first, Committable second) { + return first.checkpointId() == second.checkpointId() + && Objects.equals(first.commitMessage(), second.commitMessage()); + } + + public static boolean committableEquals(List first, List second) { + if (first.size() != second.size()) { + return false; + } + for (int i = 0; i < first.size(); i++) { + if (!committableEquals(first.get(i), second.get(i))) { + return false; + } + } + return true; + } +}