You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "fcfangcc (via GitHub)" <gi...@apache.org> on 2026/02/10 06:08:56 UTC
[PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
fcfangcc opened a new pull request, #4269:
URL: https://github.com/apache/flink-cdc/pull/4269
# Summary
This PR addresses a critical issue in the Flink CDC Iceberg sink where task interruptions or restarts during the two-phase commit (2PC) process could result in duplicate data being committed to Iceberg tables.
# Problem
In the previous implementation, the Iceberg sink did not track which Flink checkpoints had already been successfully committed to the Iceberg snapshot. If a task failed after a successful Iceberg commit but before Flink could acknowledge the checkpoint completion, or if a commit was retried, the same set of data could be committed again, leading to data inconsistency and duplication.
# Solution
We have introduced a checkpoint-id tracking mechanism inspired by standard idempotent commit patterns.
Key Changes:
Idempotent Commits in IcebergCommitter:
During the commit phase, the current Flink checkpointId is now stored in the Iceberg snapshot summary under the key
> flink-cdc-checkpoint-id
Before performing a new commit, IcebergCommitter checks the currentSnapshot of the target table. If the lastCheckpointId in the summary matches the current checkpointId, the commit is skipped with a warning, preventing double-committing the same data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4269:
URL: https://github.com/apache/flink-cdc/pull/4269#issuecomment-3881921983
Hi @fcfangcc, I would like to update the branch to optimize the current implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4269:
URL: https://github.com/apache/flink-cdc/pull/4269#discussion_r2791752251
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriter.java:
##########
@@ -87,14 +103,30 @@ public IcebergWriter(
this.taskId = taskId;
this.attemptId = attemptId;
this.zoneId = zoneId;
+ this.lastCheckpointId = lastCheckpointId;
+ this.jobId = jobId;
+ this.operatorId = operatorId;
+ LOGGER.info(
+ "IcebergWriter created, taskId: {}, attemptId: {}, lastCheckpointId: {}, jobId: {}, operatorId: {}",
+ taskId,
+ attemptId,
+ lastCheckpointId,
+ jobId,
+ operatorId);
+ }
+
+ @Override
+ public List<IcebergWriterState> snapshotState(long checkpointId) {
+ return Collections.singletonList(new IcebergWriterState(jobId, operatorId));
}
@Override
- public Collection<WriteResultWrapper> prepareCommit() throws IOException, InterruptedException {
+ public Collection<WriteResultWrapper> prepareCommit() throws IOException {
List<WriteResultWrapper> list = new ArrayList<>();
list.addAll(temporaryWriteResult);
list.addAll(getWriteResult());
temporaryWriteResult.clear();
+ lastCheckpointId++;
return list;
Review Comment:
`IcebergWriter` generates a synthetic checkpoint id by incrementing `lastCheckpointId` and using `lastCheckpointId + 1` in `getWriteResult()`, while the real Flink checkpoint id is provided to `snapshotState(long checkpointId)` but currently ignored. This can desynchronize the committable checkpointId from Flink’s checkpointing (e.g., if the runtime changes call order / retries, or if any checkpoint ids are not strictly matched 1:1 with `prepareCommit()` invocations), breaking the idempotency logic in `IcebergCommitter`.
Use the `checkpointId` argument from `snapshotState` as the source of truth (store it in a field and use it when creating `WriteResultWrapper`s), instead of maintaining your own counter via `lastCheckpointId++`.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergSink.java:
##########
@@ -57,15 +63,22 @@ public class IcebergSink
private final CompactionOptions compactionOptions;
+ private String jobId;
+
+ private String operatorId;
+
public IcebergSink(
Map<String, String> catalogOptions,
Map<String, String> tableOptions,
ZoneId zoneId,
- CompactionOptions compactionOptions) {
+ CompactionOptions compactionOptions,
+ String jobIdPrefix) {
this.catalogOptions = catalogOptions;
this.tableOptions = tableOptions;
this.zoneId = zoneId;
this.compactionOptions = compactionOptions;
+ this.jobId = jobIdPrefix + UUID.randomUUID();
+ this.operatorId = UUID.randomUUID().toString();
}
Review Comment:
The `IcebergSink` constructor signature has changed (added `jobIdPrefix`) and the previous 4-arg constructor was removed. Since `IcebergSink` is `public` and is also used directly in tests/likely by DataStream users, this is a source/binary breaking API change.
Consider adding an overloaded constructor with the previous signature that delegates to the new one using a sensible default (e.g., the existing default prefix from `IcebergDataSinkOptions.SINK_JOB_ID_PREFIX`, or a constant like `"cdc"`) to preserve backward compatibility.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4269:
URL: https://github.com/apache/flink-cdc/pull/4269#discussion_r2786171694
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergCommitter.java:
##########
@@ -56,6 +57,8 @@ public class IcebergCommitter implements Committer<WriteResultWrapper> {
public static final String TABLE_GROUP_KEY = "table";
+ public static final String CHECKPOINT_SUMMARY_NAME = "flink-cdc-checkpoint-id";
+
Review Comment:
The idempotency check uses only the Flink checkpointId stored under `flink-cdc-checkpoint-id`. Checkpoint IDs restart from the initial value when a job is started without restoring state (or when running a different job against the same table), so this can incorrectly skip legitimate new commits and cause data loss. Consider also persisting a job-scoped identifier (e.g., Flink JobID or a generated sink run UUID persisted in writer state) in the snapshot summary and require both to match before skipping.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterStateTest.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.flink.cdc.connectors.iceberg.sink.v2;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+/** A test for {@link IcebergWriterStateTest}. */
Review Comment:
The class-level Javadoc links to `IcebergWriterStateTest` (itself) rather than the class under test. This should reference `IcebergWriterState` (and/or `IcebergWriterStateSerializer`) to avoid misleading documentation.
```suggestion
/** Tests for {@link IcebergWriterState} and {@link IcebergWriterStateSerializer}. */
```
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergSink.java:
##########
@@ -92,20 +97,48 @@ public SimpleVersionedSerializer<WriteResultWrapper> getCommittableSerializer()
@Override
public SinkWriter<Event> createWriter(InitContext context) {
+ long lastCheckpointId =
+ context.getRestoredCheckpointId()
+ .orElse(CheckpointIDCounter.INITIAL_CHECKPOINT_ID - 1);
return new IcebergWriter(
catalogOptions,
context.getTaskInfo().getIndexOfThisSubtask(),
context.getTaskInfo().getAttemptNumber(),
- zoneId);
+ zoneId,
+ lastCheckpointId);
}
@Override
public SinkWriter<Event> createWriter(WriterInitContext context) {
+ long lastCheckpointId =
+ context.getRestoredCheckpointId()
+ .orElse(CheckpointIDCounter.INITIAL_CHECKPOINT_ID - 1);
return new IcebergWriter(
catalogOptions,
context.getTaskInfo().getIndexOfThisSubtask(),
context.getTaskInfo().getAttemptNumber(),
- zoneId);
+ zoneId,
+ lastCheckpointId);
+ }
+
+ @Override
+ public StatefulSinkWriter<Event, IcebergWriterState> restoreWriter(
+ WriterInitContext context, Collection<IcebergWriterState> writerStates) {
+ // No need to read checkpointId from state
+ long lastCheckpointId =
+ context.getRestoredCheckpointId()
+ .orElse(CheckpointIDCounter.INITIAL_CHECKPOINT_ID - 1);
+ return new IcebergWriter(
+ catalogOptions,
+ context.getTaskInfo().getIndexOfThisSubtask(),
+ context.getTaskInfo().getAttemptNumber(),
+ zoneId,
+ lastCheckpointId);
+ }
Review Comment:
`restoreWriter(...)` ignores the provided `writerStates`, and `IcebergWriter.snapshotState(...)` currently only stores the checkpoint id but that state is never used to restore the writer. This adds checkpointed writer state overhead without affecting behavior; either remove `SupportsWriterState/StatefulSinkWriter` entirely and rely on `getRestoredCheckpointId()`, or actually derive `lastCheckpointId` (and any future writer identity) from the restored `writerStates` for correctness and clarity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4269:
URL: https://github.com/apache/flink-cdc/pull/4269#discussion_r2792183646
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergDataSinkOptions.java:
##########
@@ -78,4 +78,12 @@ public class IcebergDataSinkOptions {
.defaultValue(-1)
.withDescription(
"The parallelism for file compaction, default value is -1, which means that compaction parallelism is equal to sink writer parallelism.");
+
+ @Experimental
+ public static final ConfigOption<String> SINK_JOB_ID_PREFIX =
+ key("sink.job.id.prefix")
Review Comment:
Thanks for your great work @fcfangcc @lvyanquan, I've minor concern about this config, IIUC, this prefix is used to distinguish the source of writing stream, and we don't have a job kind called `sink`, we can simply call this as `job.id.prefix` and the `sink` is part of a flink `job` from the concept's perspective. Could we adjust this config if possible?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan merged PR #4269:
URL: https://github.com/apache/flink-cdc/pull/4269
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4269:
URL: https://github.com/apache/flink-cdc/pull/4269#discussion_r2786848023
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergCommitter.java:
##########
@@ -56,6 +57,8 @@ public class IcebergCommitter implements Committer<WriteResultWrapper> {
public static final String TABLE_GROUP_KEY = "table";
+ public static final String CHECKPOINT_SUMMARY_NAME = "flink-cdc-checkpoint-id";
+
Review Comment:
As Copilot mentioned, using user-defined variables might introduce issues caused by other write operations executed via Flink SQL or DataStream jobs. Therefore, we could refer to the approach documented in the Iceberg repository for configuring these variables as following:
https://github.com/apache/iceberg/blob/00df4934a6b66c9f36025f45eae1fd2f7588f71f/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergCommitter.java#L290-L294
https://github.com/apache/iceberg/blob/main/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java#L163-L182
Since Flink's job ID may change after a restart, to implement this logic correctly, you also need to persist both FLINK_JOB_ID (whose initial value can be randomly generated) and OPERATOR_ID to the state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4269:
URL: https://github.com/apache/flink-cdc/pull/4269#discussion_r2791616192
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergCommitter.java:
##########
@@ -117,15 +148,47 @@ private void commit(List<WriteResultWrapper> writeResultWrappers) {
if (deleteFiles.isEmpty()) {
AppendFiles append = table.newAppend();
dataFiles.forEach(append::appendFile);
- append.commit();
+ commitOperation(append, newFlinkJobId, operatorId, checkpointId);
} else {
RowDelta delta = table.newRowDelta();
dataFiles.forEach(delta::addRows);
deleteFiles.forEach(delta::addDeletes);
- delta.commit();
+ commitOperation(delta, newFlinkJobId, operatorId, checkpointId);
+ }
+ }
+ }
+ }
+
+ private static long getMaxCommittedCheckpointId(
+ Iterable<Snapshot> ancestors, String flinkJobId, String operatorId) {
+ long lastCommittedCheckpointId = INITIAL_CHECKPOINT_ID;
+
+ for (Snapshot ancestor : ancestors) {
+ Map<String, String> summary = ancestor.summary();
+ String snapshotFlinkJobId = summary.get(SinkUtil.FLINK_JOB_ID);
+ String snapshotOperatorId = summary.get(SinkUtil.OPERATOR_ID);
+ if (flinkJobId.equals(snapshotFlinkJobId)
+ && (snapshotOperatorId == null || snapshotOperatorId.equals(operatorId))) {
+ String value = summary.get(SinkUtil.MAX_COMMITTED_CHECKPOINT_ID);
+ if (value != null) {
+ lastCommittedCheckpointId = Long.parseLong(value);
+ break;
}
}
}
+
+ return lastCommittedCheckpointId;
+ }
+
+ private static void commitOperation(
+ SnapshotUpdate<?> operation,
+ String newFlinkJobId,
+ String operatorId,
+ long checkpointId) {
+ operation.set(SinkUtil.MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
+ operation.set(SinkUtil.FLINK_JOB_ID, newFlinkJobId);
+ operation.set(SinkUtil.OPERATOR_ID, operatorId);
+ operation.commit();
Review Comment:
PR description mentions storing the checkpoint id under the key `flink-cdc-checkpoint-id`, but the implementation writes/reads Iceberg’s `SinkUtil.MAX_COMMITTED_CHECKPOINT_ID` (and `FLINK_JOB_ID`/`OPERATOR_ID`) in the snapshot summary. Please align the PR description (or the code) so users know which snapshot summary keys to look for and so the change is accurately documented.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergCommitter.java:
##########
@@ -117,15 +148,47 @@ private void commit(List<WriteResultWrapper> writeResultWrappers) {
if (deleteFiles.isEmpty()) {
AppendFiles append = table.newAppend();
dataFiles.forEach(append::appendFile);
- append.commit();
+ commitOperation(append, newFlinkJobId, operatorId, checkpointId);
} else {
RowDelta delta = table.newRowDelta();
dataFiles.forEach(delta::addRows);
deleteFiles.forEach(delta::addDeletes);
- delta.commit();
+ commitOperation(delta, newFlinkJobId, operatorId, checkpointId);
+ }
+ }
+ }
+ }
+
+ private static long getMaxCommittedCheckpointId(
+ Iterable<Snapshot> ancestors, String flinkJobId, String operatorId) {
+ long lastCommittedCheckpointId = INITIAL_CHECKPOINT_ID;
Review Comment:
`getMaxCommittedCheckpointId` initializes `lastCommittedCheckpointId` to `INITIAL_CHECKPOINT_ID`. If the table already has snapshots but none with matching `SinkUtil.FLINK_JOB_ID`/`OPERATOR_ID` (e.g., first run of this sink on an existing table), the method will return `INITIAL_CHECKPOINT_ID` and the idempotency check can incorrectly skip committing checkpoint 1, causing data loss. Initialize to `INITIAL_CHECKPOINT_ID - 1` (or another sentinel smaller than any real checkpoint) so that “no prior commit found” never matches a real checkpoint id.
```suggestion
long lastCommittedCheckpointId = INITIAL_CHECKPOINT_ID - 1;
```
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterStateSerializer.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.flink.cdc.connectors.iceberg.sink.v2;
+
+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;
+
+/** A {@link IcebergWriterStateSerializer} for {@link IcebergWriterState}. */
+public class IcebergWriterStateSerializer implements SimpleVersionedSerializer<IcebergWriterState> {
+
+ private static final int VERSION = 0;
+
+ private static final ThreadLocal<DataOutputSerializer> SERIALIZER_CACHE =
+ ThreadLocal.withInitial(() -> new DataOutputSerializer(64));
+
+ @Override
+ public int getVersion() {
+ return VERSION;
+ }
+
+ @Override
+ public byte[] serialize(IcebergWriterState icebergWriterState) throws IOException {
+ final DataOutputSerializer out = SERIALIZER_CACHE.get();
+ out.writeUTF(icebergWriterState.getJobId());
+ out.writeUTF(icebergWriterState.getOperatorId());
+ final byte[] result = out.getCopyOfBuffer();
+ out.clear();
+ return result;
+ }
+
+ @Override
+ public IcebergWriterState deserialize(int version, byte[] serialized) throws IOException {
Review Comment:
`deserialize(int version, ...)` ignores the provided `version` and will attempt to read the payload even if the version is unexpected/corrupt. For `SimpleVersionedSerializer`, it’s safer to explicitly handle supported versions (e.g., `if (version == VERSION) ... else throw new IOException("Unknown version: " + version)`) to avoid silent state corruption / confusing failures during upgrades.
```suggestion
public IcebergWriterState deserialize(int version, byte[] serialized) throws IOException {
if (version != VERSION) {
throw new IOException("Unknown version: " + version);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4269:
URL: https://github.com/apache/flink-cdc/pull/4269#discussion_r2791645319
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergCommitter.java:
##########
@@ -117,15 +148,47 @@ private void commit(List<WriteResultWrapper> writeResultWrappers) {
if (deleteFiles.isEmpty()) {
AppendFiles append = table.newAppend();
dataFiles.forEach(append::appendFile);
- append.commit();
+ commitOperation(append, newFlinkJobId, operatorId, checkpointId);
} else {
RowDelta delta = table.newRowDelta();
dataFiles.forEach(delta::addRows);
deleteFiles.forEach(delta::addDeletes);
- delta.commit();
+ commitOperation(delta, newFlinkJobId, operatorId, checkpointId);
+ }
+ }
+ }
+ }
+
+ private static long getMaxCommittedCheckpointId(
+ Iterable<Snapshot> ancestors, String flinkJobId, String operatorId) {
+ long lastCommittedCheckpointId = INITIAL_CHECKPOINT_ID;
+
+ for (Snapshot ancestor : ancestors) {
+ Map<String, String> summary = ancestor.summary();
+ String snapshotFlinkJobId = summary.get(SinkUtil.FLINK_JOB_ID);
+ String snapshotOperatorId = summary.get(SinkUtil.OPERATOR_ID);
+ if (flinkJobId.equals(snapshotFlinkJobId)
+ && (snapshotOperatorId == null || snapshotOperatorId.equals(operatorId))) {
+ String value = summary.get(SinkUtil.MAX_COMMITTED_CHECKPOINT_ID);
+ if (value != null) {
+ lastCommittedCheckpointId = Long.parseLong(value);
+ break;
}
}
}
+
+ return lastCommittedCheckpointId;
+ }
+
+ private static void commitOperation(
+ SnapshotUpdate<?> operation,
+ String newFlinkJobId,
+ String operatorId,
+ long checkpointId) {
+ operation.set(SinkUtil.MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
+ operation.set(SinkUtil.FLINK_JOB_ID, newFlinkJobId);
+ operation.set(SinkUtil.OPERATOR_ID, operatorId);
+ operation.commit();
Review Comment:
PR description was updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39056][pipeline-connector][iceberg] Duplicate Data Issue in Iceberg Sink During Two-Phase Commit [flink-cdc]
Posted by "fcfangcc (via GitHub)" <gi...@apache.org>.
fcfangcc commented on PR #4269:
URL: https://github.com/apache/flink-cdc/pull/4269#issuecomment-3881933881
> optimize the current implementation.
Ok. I don't have time for the moment. I will try when I have time It would be better if you could impl it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org