This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 7b8cbc7fef Flink: BackPort fix DynamicSink silently dropping commits
after stateless restart (#18805)
7b8cbc7fef is described below
commit 7b8cbc7fef332a8ca66cb6691c361ad0dc5963a6
Author: GuoYu <[email protected]>
AuthorDate: Thu Nov 5 17:44:21 2026 +0800
Flink: BackPort fix DynamicSink silently dropping commits after stateless
restart (#18805)
---
.../flink/sink/dynamic/DynamicCommittable.java | 24 ++-
.../sink/dynamic/DynamicCommittableSerializer.java | 24 ++-
.../flink/sink/dynamic/DynamicCommitter.java | 77 ++++---
.../flink/sink/dynamic/DynamicIcebergSink.java | 4 +-
.../sink/dynamic/DynamicWriteResultAggregator.java | 8 +-
.../iceberg/flink/sink/dynamic/JobOperatorKey.java | 12 +-
.../dynamic/TestDynamicCommittableSerializer.java | 39 +++-
.../flink/sink/dynamic/TestDynamicCommitter.java | 233 ++++++++++++++++++---
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 33 +++
.../dynamic/TestDynamicWriteResultAggregator.java | 8 +-
.../flink/sink/dynamic/DynamicCommittable.java | 24 ++-
.../sink/dynamic/DynamicCommittableSerializer.java | 24 ++-
.../flink/sink/dynamic/DynamicCommitter.java | 77 ++++---
.../flink/sink/dynamic/DynamicIcebergSink.java | 4 +-
.../sink/dynamic/DynamicWriteResultAggregator.java | 7 +-
.../iceberg/flink/sink/dynamic/JobOperatorKey.java | 12 +-
.../dynamic/TestDynamicCommittableSerializer.java | 39 +++-
.../flink/sink/dynamic/TestDynamicCommitter.java | 233 ++++++++++++++++++---
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 33 +++
.../dynamic/TestDynamicWriteResultAggregator.java | 8 +-
20 files changed, 775 insertions(+), 148 deletions(-)
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittable.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittable.java
index 4f0b68573f..1dec44e195 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittable.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittable.java
@@ -26,8 +26,10 @@ import
org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
/**
* The aggregated results of a single checkpoint which should be committed.
Containing the
- * serialized {@link DeltaManifests} files - which contains the commit data,
and the jobId,
- * operatorId, checkpointId triplet to identify the specific commit.
+ * serialized {@link DeltaManifests} files - which contains the commit data,
and the jobId, sinkId,
+ * operatorId, checkpointId to identify the specific commit.
+ *
+ * <p>Committables restored from older state use the jobId as sinkId.
*
* <p>{@link DynamicCommittableSerializer} is used to serialize {@link
DynamicCommittable} between
* the {@link DynamicWriter} and the {@link DynamicWriteResultAggregator}.
@@ -37,14 +39,21 @@ class DynamicCommittable implements Serializable {
private final TableKey key;
private final byte[][] manifests;
private final String jobId;
+ private final String sinkId;
private final String operatorId;
private final long checkpointId;
DynamicCommittable(
- TableKey key, byte[][] manifests, String jobId, String operatorId, long
checkpointId) {
+ TableKey key,
+ byte[][] manifests,
+ String jobId,
+ String sinkId,
+ String operatorId,
+ long checkpointId) {
this.key = key;
this.manifests = manifests;
this.jobId = jobId;
+ this.sinkId = sinkId;
this.operatorId = operatorId;
this.checkpointId = checkpointId;
}
@@ -61,6 +70,10 @@ class DynamicCommittable implements Serializable {
return jobId;
}
+ String sinkId() {
+ return sinkId;
+ }
+
String operatorId() {
return operatorId;
}
@@ -80,12 +93,14 @@ class DynamicCommittable implements Serializable {
&& Objects.equals(key, that.key)
&& Arrays.deepEquals(manifests, that.manifests)
&& Objects.equals(jobId, that.jobId)
+ && Objects.equals(sinkId, that.sinkId)
&& Objects.equals(operatorId, that.operatorId);
}
@Override
public int hashCode() {
- return Objects.hash(key, Arrays.deepHashCode(manifests), jobId,
operatorId, checkpointId);
+ return Objects.hash(
+ key, Arrays.deepHashCode(manifests), jobId, sinkId, operatorId,
checkpointId);
}
@Override
@@ -93,6 +108,7 @@ class DynamicCommittable implements Serializable {
return MoreObjects.toStringHelper(this)
.add("key", key)
.add("jobId", jobId)
+ .add("sinkId", sinkId)
.add("checkpointId", checkpointId)
.add("operatorId", operatorId)
.toString();
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java
index d599d29dba..e4f3808cd8 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java
@@ -33,10 +33,11 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
private static final int VERSION_1 = 1;
private static final int VERSION_2 = 2;
+ private static final int VERSION_3 = 3;
@Override
public int getVersion() {
- return VERSION_2;
+ return VERSION_3;
}
@Override
@@ -45,6 +46,7 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
committable.key().serializeTo(view);
view.writeUTF(committable.jobId());
+ view.writeUTF(committable.sinkId());
view.writeUTF(committable.operatorId());
view.writeLong(committable.checkpointId());
@@ -65,6 +67,8 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
return deserializeV1(serialized);
} else if (version == VERSION_2) {
return deserializeV2(serialized);
+ } else if (version == VERSION_3) {
+ return deserializeV3(serialized);
}
throw new IOException("Unrecognized version or corrupt state: " + version);
@@ -83,6 +87,7 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
new TableKey(key.tableName(), key.branch()),
new byte[][] {manifestBuf},
jobId,
+ jobId,
operatorId,
checkpointId);
}
@@ -93,7 +98,22 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
String jobId = view.readUTF();
String operatorId = view.readUTF();
long checkpointId = view.readLong();
+ byte[][] manifests = readManifests(view);
+ return new DynamicCommittable(key, manifests, jobId, jobId, operatorId,
checkpointId);
+ }
+
+ private DynamicCommittable deserializeV3(byte[] serialized) throws
IOException {
+ DataInputDeserializer view = new DataInputDeserializer(serialized);
+ TableKey key = TableKey.deserializeFrom(view);
+ String jobId = view.readUTF();
+ String sinkId = view.readUTF();
+ String operatorId = view.readUTF();
+ long checkpointId = view.readLong();
+ byte[][] manifests = readManifests(view);
+ return new DynamicCommittable(key, manifests, jobId, sinkId, operatorId,
checkpointId);
+ }
+ private static byte[][] readManifests(DataInputDeserializer view) throws
IOException {
byte[][] manifestsBuf = new byte[view.readInt()][];
for (int i = 0; i < manifestsBuf.length; i++) {
byte[] manifest = new byte[view.readInt()];
@@ -101,6 +121,6 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
manifestsBuf[i] = manifest;
}
- return new DynamicCommittable(key, manifestsBuf, jobId, operatorId,
checkpointId);
+ return manifestsBuf;
}
}
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
index c8b9a097ad..7a4bdf86aa 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
@@ -70,7 +70,7 @@ import org.slf4j.LoggerFactory;
* <li>There is no late checkpoint - if checkpoint 'x' has received in one
call, then after a
* successful run only checkpoints > x will arrive
* <li>There is no other writer which would generate another commit to the
same branch with the
- * same jobId-operatorId-checkpointId triplet
+ * same sinkId-operatorId-checkpointId triplet
* </ul>
*/
@Internal
@@ -82,6 +82,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
private static final long INITIAL_CHECKPOINT_ID = -1L;
private static final String FLINK_JOB_ID = "flink.job-id";
+ private static final String SINK_ID = "flink.sink-id";
private static final String OPERATOR_ID = "flink.operator-id";
private final Map<String, String> snapshotProperties;
private final boolean replacePartitions;
@@ -118,9 +119,8 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
| | | |
| | | +-- commit
requests at that checkpoint
| | +-- checkpointId, sorted
ascending so older commits go first
- | +-- (jobId, operatorId) of the producing
aggregator; deduplication against the
- | table's snapshot summaries is per (jobId,
operatorId), so each group is
- | walked against the ancestor chain
independently
+ | +-- (sinkId, operatorId) of the producing
aggregator; each group is
+ | deduplicated against the ancestor chain
independently
+-- (tableName, branch); we load the table and its ancestor
chain once per TableKey
Each (table, branch, checkpoint) triplet must have only one commit
request.
@@ -156,17 +156,17 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
List<Map.Entry<JobOperatorKey, NavigableMap<Long,
List<CommitRequest<DynamicCommittable>>>>>
jobEntries = Lists.newArrayList(tableEntry.getValue().entrySet());
- // Preserve checkpoint order across groups so that older-jobId commits
land before newer-jobId
- // ones when the batch mixes committables from different jobIds (e.g.
state replay after a
- // restart). Within a (jobId, operatorId) group, checkpoint order is
already guaranteed by
- // the inner NavigableMap.
+ // Preserve checkpoint order across groups so that older-sinkId commits
land before
+ // newer-sinkId ones when the batch mixes committables from different
runs (e.g. state replay
+ // after a restart). Within a (sinkId, operatorId) group, checkpoint
order is already
+ // guaranteed by the inner NavigableMap.
jobEntries.sort(Comparator.comparingLong(entry ->
entry.getValue().firstKey()));
for (Map.Entry<JobOperatorKey, NavigableMap<Long,
List<CommitRequest<DynamicCommittable>>>>
jobEntry : jobEntries) {
JobOperatorKey jobKey = jobEntry.getKey();
long maxCommittedCheckpointId =
- getMaxCommittedCheckpointId(ancestors, jobKey.jobId(),
jobKey.operatorId());
+ getMaxCommittedCheckpointId(ancestors, jobKey.sinkId(),
jobKey.operatorId());
NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>
skippedCommitRequests =
jobEntry.getValue().headMap(maxCommittedCheckpointId, true);
@@ -181,22 +181,24 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
jobEntry.getValue().tailMap(maxCommittedCheckpointId, false);
if (!uncommitted.isEmpty()) {
commitPendingRequests(
- table, tableKey.branch(), uncommitted, jobKey.jobId(),
jobKey.operatorId());
+ table,
+ tableKey.branch(),
+ uncommitted,
+ jobKey.jobId(),
+ jobKey.sinkId(),
+ jobKey.operatorId());
}
}
}
}
private static long getMaxCommittedCheckpointId(
- Iterable<Snapshot> ancestors, String flinkJobId, String operatorId) {
+ Iterable<Snapshot> ancestors, String sinkId, String operatorId) {
long lastCommittedCheckpointId = INITIAL_CHECKPOINT_ID;
for (Snapshot ancestor : ancestors) {
Map<String, String> summary = ancestor.summary();
- String snapshotFlinkJobId = summary.get(FLINK_JOB_ID);
- String snapshotOperatorId = summary.get(OPERATOR_ID);
- if (flinkJobId.equals(snapshotFlinkJobId)
- && (snapshotOperatorId == null ||
snapshotOperatorId.equals(operatorId))) {
+ if (writtenBySameSink(summary, sinkId, operatorId)) {
String value = summary.get(MAX_COMMITTED_CHECKPOINT_ID);
if (value != null) {
lastCommittedCheckpointId = Long.parseLong(value);
@@ -208,6 +210,18 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
return lastCommittedCheckpointId;
}
+ private static boolean writtenBySameSink(
+ Map<String, String> summary, String sinkId, String operatorId) {
+ String snapshotOperatorId = summary.get(OPERATOR_ID);
+ if (snapshotOperatorId != null && !snapshotOperatorId.equals(operatorId)) {
+ return false;
+ }
+
+ // Snapshots written before the sinkId was introduced are identified by
the jobId
+ String snapshotSinkId = summary.get(SINK_ID);
+ return sinkId.equals(snapshotSinkId != null ? snapshotSinkId :
summary.get(FLINK_JOB_ID));
+ }
+
/**
* Commits the data to the Iceberg table by reading the file data from the
{@link DeltaManifests}
* ordered by the checkpointId, and writing the new snapshot to the Iceberg
table. The {@link
@@ -216,6 +230,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
*
* @param commitRequestMap The checkpointId to {@link CommitRequest} map of
the changes to commit
* @param newFlinkJobId The jobId to store in the {@link SnapshotSummary}
+ * @param sinkId The sinkId to store in the {@link SnapshotSummary}
* @param operatorId The operatorId to store in the {@link SnapshotSummary}
* @throws IOException On commit failure
*/
@@ -224,6 +239,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
String branch,
NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>
commitRequestMap,
String newFlinkJobId,
+ String sinkId,
String operatorId)
throws IOException {
long checkpointId = commitRequestMap.lastKey();
@@ -258,9 +274,9 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
}
if (replacePartitions) {
- replacePartitions(table, branch, pendingResults, newFlinkJobId,
operatorId);
+ replacePartitions(table, branch, pendingResults, newFlinkJobId, sinkId,
operatorId);
} else {
- commitDeltaTxn(table, branch, pendingResults, newFlinkJobId, operatorId);
+ commitDeltaTxn(table, branch, pendingResults, newFlinkJobId, sinkId,
operatorId);
}
FlinkManifestUtil.deleteCommittedManifests(table, manifests,
newFlinkJobId, checkpointId);
@@ -271,6 +287,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
String branch,
NavigableMap<Long, List<WriteResult>> pendingResults,
String newFlinkJobId,
+ String sinkId,
String operatorId) {
// Iceberg tables are unsorted. So the order of the append data does not
matter.
// Hence, we commit everything in one snapshot.
@@ -292,6 +309,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
summary,
"dynamic partition overwrite",
newFlinkJobId,
+ sinkId,
operatorId,
pendingResults.lastKey());
}
@@ -301,6 +319,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
String branch,
NavigableMap<Long, List<WriteResult>> pendingResults,
String newFlinkJobId,
+ String sinkId,
String operatorId) {
for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
long checkpointId = e.getKey();
@@ -328,7 +347,15 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
// pending checkpoints here are very rare to occur, i.e. only with very
short checkpoint
// intervals or when concurrent checkpointing is enabled.
commitOperation(
- table, branch, rowDelta, summary, "rowDelta", newFlinkJobId,
operatorId, checkpointId);
+ table,
+ branch,
+ rowDelta,
+ summary,
+ "rowDelta",
+ newFlinkJobId,
+ sinkId,
+ operatorId,
+ checkpointId);
}
}
@@ -340,20 +367,20 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
private static class MaxCommittedCheckpointIdValidator implements
SnapshotAncestryValidator {
private final long stagedCheckpointId;
- private final String flinkJobId;
+ private final String sinkId;
private final String flinkOperatorId;
private MaxCommittedCheckpointIdValidator(
- long stagedCheckpointId, String flinkJobId, String flinkOperatorId) {
+ long stagedCheckpointId, String sinkId, String flinkOperatorId) {
this.stagedCheckpointId = stagedCheckpointId;
- this.flinkJobId = flinkJobId;
+ this.sinkId = sinkId;
this.flinkOperatorId = flinkOperatorId;
}
@Override
public boolean validate(Iterable<Snapshot> baseSnapshots) {
long maxCommittedCheckpointId =
- getMaxCommittedCheckpointId(baseSnapshots, flinkJobId,
flinkOperatorId);
+ getMaxCommittedCheckpointId(baseSnapshots, sinkId, flinkOperatorId);
if (maxCommittedCheckpointId >= stagedCheckpointId) {
throw new MaxCommittedCheckpointMismatchException();
}
@@ -370,6 +397,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
CommitSummary summary,
String description,
String newFlinkJobId,
+ String sinkId,
String operatorId,
long checkpointId) {
@@ -385,10 +413,11 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
// used by the sink.
operation.set(MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
operation.set(FLINK_JOB_ID, newFlinkJobId);
+ operation.set(SINK_ID, sinkId);
+
operation.set(OPERATOR_ID, operatorId);
operation.toBranch(branch);
- operation.validateWith(
- new MaxCommittedCheckpointIdValidator(checkpointId, newFlinkJobId,
operatorId));
+ operation.validateWith(new MaxCommittedCheckpointIdValidator(checkpointId,
sinkId, operatorId));
long startNano = System.nanoTime();
try {
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
index 42c52894ad..652e9a409d 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
@@ -99,7 +99,7 @@ public class DynamicIcebergSink
this.cacheMaximumSize = cacheMaximumSize;
// We generate a random UUID every time when a sink is created.
// This is used to separate files generated by different sinks writing the
same table.
- // Also used to generate the aggregator operator name
+ // Also used to generate the aggregator operator name, and to deduplicate
commits
this.sinkId = UUID.randomUUID().toString();
this.forwardWriteResults = forwardWriteResults;
}
@@ -169,7 +169,7 @@ public class DynamicIcebergSink
.transform(
prefixIfNotNull(uidPrefix, sinkId + " Pre Commit"),
typeInformation,
- new DynamicWriteResultAggregator(catalogLoader, cacheMaximumSize))
+ new DynamicWriteResultAggregator(catalogLoader, cacheMaximumSize,
sinkId))
.uid(prefixIfNotNull(uidPrefix, "-pre-commit-topology"));
}
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java
index 47d239200b..fd07c9c1d5 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java
@@ -50,7 +50,7 @@ import org.slf4j.LoggerFactory;
/**
* Operator which aggregates the individual {@link WriteResult} objects to a
single {@link
- * DynamicCommittable} per checkpoint (storing the serialized {@link
DeltaManifests}, jobId,
+ * DynamicCommittable} per checkpoint (storing the serialized {@link
DeltaManifests}, jobId, sinkId,
* operatorId, checkpointId)
*/
class DynamicWriteResultAggregator
@@ -61,6 +61,8 @@ class DynamicWriteResultAggregator
private final CatalogLoader catalogLoader;
private final int cacheMaximumSize;
+ private final String sinkId;
+
private transient Map<TableKey, Map<Integer, Collection<WriteResult>>>
resultsByTableKeyAndSpec;
private transient Map<String, Map<Integer, PartitionSpec>> specs;
private transient Map<String, Tuple2<ManifestOutputFileFactory, Integer>>
@@ -71,9 +73,10 @@ class DynamicWriteResultAggregator
private transient int attemptId;
private transient Catalog catalog;
- DynamicWriteResultAggregator(CatalogLoader catalogLoader, int
cacheMaximumSize) {
+ DynamicWriteResultAggregator(CatalogLoader catalogLoader, int
cacheMaximumSize, String sinkId) {
this.catalogLoader = catalogLoader;
this.cacheMaximumSize = cacheMaximumSize;
+ this.sinkId = sinkId;
}
@Override
@@ -106,6 +109,7 @@ class DynamicWriteResultAggregator
entries.getKey(),
writeToManifests(entries.getKey().tableName(),
entries.getValue(), checkpointId),
getContainingTask().getEnvironment().getJobID().toString(),
+ sinkId,
getRuntimeContext().getOperatorUniqueID(),
checkpointId),
checkpointId,
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java
index 463c09475d..097f31b7ba 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java
@@ -22,10 +22,12 @@ import java.util.Objects;
class JobOperatorKey {
private final String jobId;
+ private final String sinkId;
private final String operatorId;
JobOperatorKey(DynamicCommittable committable) {
this.jobId = committable.jobId();
+ this.sinkId = committable.sinkId();
this.operatorId = committable.operatorId();
}
@@ -33,6 +35,10 @@ class JobOperatorKey {
return jobId;
}
+ String sinkId() {
+ return sinkId;
+ }
+
String operatorId() {
return operatorId;
}
@@ -47,11 +53,13 @@ class JobOperatorKey {
return false;
}
- return Objects.equals(jobId, that.jobId) && Objects.equals(operatorId,
that.operatorId);
+ return Objects.equals(jobId, that.jobId)
+ && Objects.equals(sinkId, that.sinkId)
+ && Objects.equals(operatorId, that.operatorId);
}
@Override
public int hashCode() {
- return Objects.hash(jobId, operatorId);
+ return Objects.hash(jobId, sinkId, operatorId);
}
}
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommittableSerializer.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommittableSerializer.java
index 16890d1f63..55da85b2e1 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommittableSerializer.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommittableSerializer.java
@@ -24,6 +24,7 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Set;
+import java.util.UUID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.runtime.jobgraph.OperatorID;
@@ -36,22 +37,42 @@ class TestDynamicCommittableSerializer {
new TableKey("table", "branch"),
new byte[][] {{3, 4}, {5, 6}},
JobID.generate().toHexString(),
+ UUID.randomUUID().toString(),
new OperatorID().toHexString(),
5);
@Test
void testV1() throws IOException {
+ String jobId = JobID.generate().toHexString();
+ // Older formats have no sinkId, the jobId is used instead
var committable =
new DynamicCommittable(
new TableKey("table", "branch"),
new byte[][] {{3, 4}},
- JobID.generate().toHexString(),
+ jobId,
+ jobId,
new OperatorID().toHexString(),
5);
DynamicCommittableSerializer serializer = new
DynamicCommittableSerializer();
assertThat(serializer.deserialize(1,
serializeV1(committable))).isEqualTo(committable);
}
+ @Test
+ void testV2() throws IOException {
+ String jobId = JobID.generate().toHexString();
+ // Older formats have no sinkId, the jobId is used instead
+ var committable =
+ new DynamicCommittable(
+ new TableKey("table", "branch"),
+ new byte[][] {{3, 4}, {5, 6}},
+ jobId,
+ jobId,
+ new OperatorID().toHexString(),
+ 5);
+ DynamicCommittableSerializer serializer = new
DynamicCommittableSerializer();
+ assertThat(serializer.deserialize(2,
serializeV2(committable))).isEqualTo(committable);
+ }
+
@Test
void testLatestVersion() throws IOException {
DynamicCommittableSerializer serializer = new
DynamicCommittableSerializer();
@@ -67,6 +88,22 @@ class TestDynamicCommittableSerializer {
.isInstanceOf(IOException.class);
}
+ byte[] serializeV2(DynamicCommittable committable) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
+ committable.key().serializeTo(view);
+ view.writeUTF(committable.jobId());
+ view.writeUTF(committable.operatorId());
+ view.writeLong(committable.checkpointId());
+ view.writeInt(committable.manifests().length);
+ for (byte[] manifest : committable.manifests()) {
+ view.writeInt(manifest.length);
+ view.write(manifest);
+ }
+
+ return out.toByteArray();
+ }
+
byte[] serializeV1(DynamicCommittable committable) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java
index aecad67c1a..0133879ff1 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java
@@ -27,6 +27,7 @@ import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Map;
+import java.util.UUID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.connector.sink2.Committer.CommitRequest;
import org.apache.flink.api.connector.sink2.mocks.MockCommitRequest;
@@ -67,6 +68,7 @@ import org.junit.jupiter.params.provider.ValueSource;
class TestDynamicCommitter {
static final String DB = "db";
+ static final String SINK_ID = "sinkId";
static final String TABLE1 = "table";
static final String TABLE2 = "table2";
@@ -170,7 +172,8 @@ class TestDynamicCommitter {
TableKey tableKey3 = new TableKey(TABLE2, "branch2");
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -188,15 +191,18 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey1, deltaManifests1, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey1, deltaManifests1, jobId, SINK_ID, operatorId,
checkpointId));
CommitRequest<DynamicCommittable> commitRequest2 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey2, deltaManifests2, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey2, deltaManifests2, jobId, SINK_ID, operatorId,
checkpointId));
CommitRequest<DynamicCommittable> commitRequest3 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey3, deltaManifests3, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey3, deltaManifests3, jobId, SINK_ID, operatorId,
checkpointId));
dynamicCommitter.commit(Sets.newHashSet(commitRequest1, commitRequest2,
commitRequest3));
@@ -210,6 +216,7 @@ class TestDynamicCommitter {
.put("added-records", "42")
.put("changed-partition-count", "1")
.put("flink.job-id", jobId)
+ .put("flink.sink-id", SINK_ID)
.put("flink.max-committed-checkpoint-id", "" + checkpointId)
.put("flink.operator-id", operatorId)
.put("total-data-files", "1")
@@ -227,6 +234,7 @@ class TestDynamicCommitter {
.put("added-records", "42")
.put("changed-partition-count", "1")
.put("flink.job-id", jobId)
+ .put("flink.sink-id", SINK_ID)
.put("flink.max-committed-checkpoint-id", "" + checkpointId)
.put("flink.operator-id", operatorId)
.put("total-data-files", "1")
@@ -247,6 +255,7 @@ class TestDynamicCommitter {
.put("added-records", "42")
.put("changed-partition-count", "1")
.put("flink.job-id", jobId)
+ .put("flink.sink-id", SINK_ID)
.put("flink.max-committed-checkpoint-id", "" + checkpointId)
.put("flink.operator-id", operatorId)
.put("total-data-files", "1")
@@ -280,7 +289,8 @@ class TestDynamicCommitter {
TableKey tableKey = new TableKey(TABLE1, "branch");
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -294,13 +304,15 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId));
dynamicCommitter.commit(Sets.newHashSet(commitRequest));
CommitRequest<DynamicCommittable> oldCommitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId - 1));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId - 1));
// Old commits requests shouldn't affect the result
dynamicCommitter.commit(Sets.newHashSet(oldCommitRequest));
@@ -347,7 +359,8 @@ class TestDynamicCommitter {
previousCommitterMetrics);
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -356,6 +369,7 @@ class TestDynamicCommitter {
// Operator id is stable across Flink job restarts, jobId is not.
final String operatorId = new OperatorID().toHexString();
final String previousJobIdStr = previousJobId.toHexString();
+ final String previousSinkId = UUID.randomUUID().toString();
final int previousCheckpointId = 10;
byte[][] previousManifests =
@@ -364,7 +378,12 @@ class TestDynamicCommitter {
DynamicCommittable previousCommittable =
new DynamicCommittable(
- tableKey, previousManifests, previousJobIdStr, operatorId,
previousCheckpointId);
+ tableKey,
+ previousManifests,
+ previousJobIdStr,
+ previousSinkId,
+ operatorId,
+ previousCheckpointId);
previousCommitter.commit(Sets.newHashSet(new
MockCommitRequest<>(previousCommittable)));
table.refresh();
@@ -382,6 +401,7 @@ class TestDynamicCommitter {
newCommitterMetrics);
final String newJobIdStr = newJobId.toHexString();
+ final String newSinkId = UUID.randomUUID().toString();
final int newCheckpointId = previousCheckpointId + 1;
byte[][] previousManifestsNew =
@@ -396,12 +416,13 @@ class TestDynamicCommitter {
tableKey,
previousManifestsNew,
previousJobIdStr,
+ previousSinkId,
operatorId,
previousCheckpointId));
CommitRequest<DynamicCommittable> newCommitRequest =
new MockCommitRequest<>(
new DynamicCommittable(
- tableKey, newManifests, newJobIdStr, operatorId,
newCheckpointId));
+ tableKey, newManifests, newJobIdStr, newSinkId, operatorId,
newCheckpointId));
newCommitter.commit(Sets.newHashSet(replayedPreviousCommitRequest,
newCommitRequest));
@@ -411,12 +432,14 @@ class TestDynamicCommitter {
Snapshot first = Iterables.get(table.snapshots(), 0);
assertThat(first.summary())
.containsEntry("flink.job-id", previousJobIdStr)
+ .containsEntry("flink.sink-id", previousSinkId)
.containsEntry("flink.max-committed-checkpoint-id",
String.valueOf(previousCheckpointId))
.containsEntry("flink.operator-id", operatorId);
Snapshot second = Iterables.get(table.snapshots(), 1);
assertThat(second.summary())
.containsEntry("flink.job-id", newJobIdStr)
+ .containsEntry("flink.sink-id", newSinkId)
.containsEntry("flink.max-committed-checkpoint-id",
String.valueOf(newCheckpointId))
.containsEntry("flink.operator-id", operatorId);
}
@@ -441,7 +464,8 @@ class TestDynamicCommitter {
committerMetrics);
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -460,10 +484,12 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> oldRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, oldManifests, oldJobId,
operatorId, oldCheckpointId));
+ new DynamicCommittable(
+ tableKey, oldManifests, oldJobId, oldJobId, operatorId,
oldCheckpointId));
CommitRequest<DynamicCommittable> newRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, newManifests, newJobId,
operatorId, newCheckpointId));
+ new DynamicCommittable(
+ tableKey, newManifests, newJobId, newJobId, operatorId,
newCheckpointId));
// Hand the requests in reversed order; the committer must still land them
in checkpointId
// order on the snapshot chain.
@@ -506,7 +532,8 @@ class TestDynamicCommitter {
committerMetrics);
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -530,7 +557,8 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId));
// Upgrade the table version
UpdateProperties updateApi = table1.updateProperties();
@@ -567,7 +595,8 @@ class TestDynamicCommitter {
TableKey tableKey = new TableKey(TABLE1, "branch");
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -581,7 +610,8 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId));
dynamicCommitter.commit(Sets.newHashSet(commitRequest));
@@ -618,7 +648,8 @@ class TestDynamicCommitter {
assertThat(table.snapshots()).isEmpty();
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -636,21 +667,24 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey1, deltaManifests1, jobId,
operatorId, checkpointId1));
+ new DynamicCommittable(
+ tableKey1, deltaManifests1, jobId, jobId, operatorId,
checkpointId1));
byte[][] deltaManifests2 =
aggregator.writeToManifests(tableKey1.tableName(),
WRITE_RESULT_BY_SPEC_2, checkpointId1);
CommitRequest<DynamicCommittable> commitRequest2 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey1, deltaManifests2, jobId,
operatorId, checkpointId1));
+ new DynamicCommittable(
+ tableKey1, deltaManifests2, jobId, jobId, operatorId,
checkpointId1));
byte[][] deltaManifests3 =
aggregator.writeToManifests(tableKey2.tableName(),
WRITE_RESULT_BY_SPEC_2, checkpointId2);
CommitRequest<DynamicCommittable> commitRequest3 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey2, deltaManifests3, jobId,
operatorId, checkpointId2));
+ new DynamicCommittable(
+ tableKey2, deltaManifests3, jobId, jobId, operatorId,
checkpointId2));
boolean overwriteMode = false;
int workerPoolSize = 1;
@@ -715,7 +749,8 @@ class TestDynamicCommitter {
assertThat(table.snapshots()).isEmpty();
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -741,15 +776,18 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests1, jobId,
operatorId, checkpointId1));
+ new DynamicCommittable(
+ tableKey, deltaManifests1, jobId, jobId, operatorId,
checkpointId1));
CommitRequest<DynamicCommittable> commitRequest2 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests2, jobId,
operatorId, checkpointId2));
+ new DynamicCommittable(
+ tableKey, deltaManifests2, jobId, jobId, operatorId,
checkpointId2));
CommitRequest<DynamicCommittable> commitRequest3 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests3, jobId,
operatorId, checkpointId3));
+ new DynamicCommittable(
+ tableKey, deltaManifests3, jobId, jobId, operatorId,
checkpointId3));
boolean overwriteMode = false;
int workerPoolSize = 1;
@@ -860,7 +898,8 @@ class TestDynamicCommitter {
assertThat(table.snapshots()).isEmpty();
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -875,14 +914,16 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifest1, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifest1, jobId, jobId, operatorId,
checkpointId));
byte[][] deltaManifest2 =
aggregator.writeToManifests(tableKey.tableName(),
WRITE_RESULT_BY_SPEC_2, checkpointId);
CommitRequest<DynamicCommittable> commitRequest2 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifest2, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifest2, jobId, jobId, operatorId,
checkpointId));
boolean overwriteMode = false;
int workerPoolSize = 1;
@@ -930,7 +971,8 @@ class TestDynamicCommitter {
TableKey tableKey = new TableKey(TABLE1, "branch");
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -944,7 +986,8 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId));
dynamicCommitter.commit(Sets.newHashSet(commitRequest));
@@ -954,7 +997,7 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> overwriteRequest =
new MockCommitRequest<>(
new DynamicCommittable(
- tableKey, overwriteManifests, jobId, operatorId, checkpointId
+ 1));
+ tableKey, overwriteManifests, jobId, jobId, operatorId,
checkpointId + 1));
dynamicCommitter.commit(Sets.newHashSet(overwriteRequest));
@@ -987,7 +1030,8 @@ class TestDynamicCommitter {
assertThat(table.snapshots()).isEmpty();
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -1003,7 +1047,7 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, manifests, jobId, operatorId,
checkpointId));
+ new DynamicCommittable(tableKey, manifests, jobId, jobId,
operatorId, checkpointId));
Collection<CommitRequest<DynamicCommittable>> commitRequests =
Sets.newHashSet(commitRequest1);
int workerPoolSize = 1;
@@ -1055,6 +1099,118 @@ class TestDynamicCommitter {
.build());
}
+ @Test
+ void testCommitAfterStatelessRestartWithSameJobId() throws Exception {
+ Table table = catalog.loadTable(TableIdentifier.of(TABLE1));
+ DynamicWriteResultAggregator aggregator = openAggregator();
+ TableKey tableKey = new TableKey(TABLE1, "branch");
+ String jobId = JobID.generate().toHexString();
+ String operatorId = new OperatorID().toHexString();
+ String newSinkId = UUID.randomUUID().toString();
+
+ createCommitter()
+ .commit(
+ Sets.newHashSet(
+ commitRequest(
+ aggregator,
+ tableKey,
+ WRITE_RESULT_BY_SPEC,
+ jobId,
+ UUID.randomUUID().toString(),
+ operatorId,
+ 10)));
+
+ createCommitter()
+ .commit(
+ Sets.newHashSet(
+ commitRequest(
+ aggregator,
+ tableKey,
+ WRITE_RESULT_BY_SPEC_2,
+ jobId,
+ newSinkId,
+ operatorId,
+ 1)));
+
+ table.refresh();
+ assertThat(table.snapshots()).hasSize(2);
+
assertThat(table.snapshot(table.refs().get("branch").snapshotId()).summary())
+ .containsEntry("flink.job-id", jobId)
+ .containsEntry("flink.sink-id", newSinkId)
+ .containsEntry("flink.max-committed-checkpoint-id", "1")
+ .containsEntry("total-records", "66");
+ }
+
+ @Test
+ void testSkipsAlreadyCommittedOlderCommittable() throws Exception {
+ Table table = catalog.loadTable(TableIdentifier.of(TABLE1));
+ DynamicWriteResultAggregator aggregator = openAggregator();
+ TableKey tableKey = new TableKey(TABLE1, "branch");
+ String jobId = JobID.generate().toHexString();
+ String operatorId = new OperatorID().toHexString();
+
+ // Snapshot written before the sinkId was introduced
+ table
+ .newAppend()
+ .appendFile(DATA_FILE)
+ .set("flink.job-id", jobId)
+ .set("flink.operator-id", operatorId)
+ .set("flink.max-committed-checkpoint-id", "5")
+ .toBranch("branch")
+ .commit();
+
+ // Committables from older state use the jobId as sinkId
+ createCommitter()
+ .commit(
+ Sets.newHashSet(
+ commitRequest(
+ aggregator, tableKey, WRITE_RESULT_BY_SPEC, jobId, jobId,
operatorId, 5),
+ commitRequest(
+ aggregator, tableKey, WRITE_RESULT_BY_SPEC_2, jobId,
jobId, operatorId, 6)));
+
+ table.refresh();
+ assertThat(table.snapshots()).hasSize(2);
+
assertThat(table.snapshot(table.refs().get("branch").snapshotId()).summary())
+ .containsEntry("flink.sink-id", jobId)
+ .containsEntry("flink.max-committed-checkpoint-id", "6")
+ .containsEntry("total-records", "66");
+ }
+
+ private DynamicWriteResultAggregator openAggregator() throws Exception {
+ DynamicWriteResultAggregator aggregator =
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, SINK_ID);
+ OneInputStreamOperatorTestHarness aggregatorHarness =
+ new OneInputStreamOperatorTestHarness(aggregator);
+ aggregatorHarness.open();
+ return aggregator;
+ }
+
+ private static DynamicCommitter createCommitter() {
+ return new DynamicCommitter(
+ CATALOG_EXTENSION.catalog(),
+ Maps.newHashMap(),
+ false,
+ 1,
+ "sinkId",
+ new DynamicCommitterMetrics(new UnregisteredMetricsGroup()));
+ }
+
+ private static CommitRequest<DynamicCommittable> commitRequest(
+ DynamicWriteResultAggregator aggregator,
+ TableKey tableKey,
+ Map<Integer, Collection<WriteResult>> writeResults,
+ String jobId,
+ String sinkId,
+ String operatorId,
+ long checkpointId)
+ throws IOException {
+ byte[][] manifests =
+ aggregator.writeToManifests(tableKey.tableName(), writeResults,
checkpointId);
+ return new MockCommitRequest<>(
+ new DynamicCommittable(tableKey, manifests, jobId, sinkId, operatorId,
checkpointId));
+ }
+
interface CommitHook extends Serializable {
default Collection<CommitRequest<DynamicCommittable>> beforeCommit(
Collection<CommitRequest<DynamicCommittable>> commitRequests) {
@@ -1153,11 +1309,20 @@ class TestDynamicCommitter {
CommitSummary summary,
String description,
String newFlinkJobId,
+ String sinkId,
String operatorId,
long checkpointId) {
commitHook.beforeCommitOperation();
super.commitOperation(
- table, branch, operation, summary, description, newFlinkJobId,
operatorId, checkpointId);
+ table,
+ branch,
+ operation,
+ summary,
+ description,
+ newFlinkJobId,
+ sinkId,
+ operatorId,
+ checkpointId);
commitHook.afterCommitOperation();
}
}
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 873ef2475f..b7b0e53313 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -306,6 +306,37 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
runTest(rows);
}
+ @Test
+ void testAllTablesShareTheSinkId() throws Exception {
+ List<String> tableNames = List.of("t1", "t2", "t3", "t4");
+ List<DynamicIcebergDataImpl> rows = Lists.newArrayList();
+ for (String tableName : tableNames) {
+ rows.add(
+ new DynamicIcebergDataImpl(
+ SimpleDataUtil.SCHEMA,
+ tableName,
+ SnapshotRef.MAIN_BRANCH,
+ PartitionSpec.unpartitioned()));
+ }
+
+ executeDynamicSink(rows, env, true, 2, null);
+
+ Set<String> sinkIds = Sets.newHashSet();
+ for (String tableName : tableNames) {
+ Map<String, String> summary =
+ CATALOG_EXTENSION
+ .catalog()
+ .loadTable(TableIdentifier.of(DATABASE, tableName))
+ .currentSnapshot()
+ .summary();
+ assertThat(summary).containsKey("flink.sink-id");
+
assertThat(summary.get("flink.sink-id")).isNotEqualTo(summary.get("flink.job-id"));
+ sinkIds.add(summary.get("flink.sink-id"));
+ }
+
+ assertThat(sinkIds).hasSize(1);
+ }
+
@Test
void testNoShuffleTopology() throws Exception {
DataStream<DynamicIcebergDataImpl> dataStream =
@@ -1697,11 +1728,13 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
.toBranch(original.key().branch())
.commit();
+ // Committables restored from older state use the jobId as sinkId
DynamicCommittable replayed =
new DynamicCommittable(
original.key(),
original.manifests(),
PREVIOUS_JOB_ID,
+ PREVIOUS_JOB_ID,
original.operatorId(),
replayedCheckpointId);
List<Committer.CommitRequest<DynamicCommittable>> enriched =
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java
index f68e8dae17..3786cea425 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java
@@ -64,7 +64,8 @@ class TestDynamicWriteResultAggregator {
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table2"), new
Schema());
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
try (OneInputStreamOperatorTestHarness<
CommittableMessage<DynamicWriteResult>,
CommittableMessage<DynamicCommittable>>
testHarness = new OneInputStreamOperatorTestHarness<>(aggregator)) {
@@ -107,7 +108,8 @@ class TestDynamicWriteResultAggregator {
// Disable caching of ManifestOutputFileFactory.
final int zeroCacheSize = 0;
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
zeroCacheSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), zeroCacheSize, "sinkId");
try (OneInputStreamOperatorTestHarness<
CommittableMessage<DynamicWriteResult>,
CommittableMessage<DynamicCommittable>>
testHarness = new OneInputStreamOperatorTestHarness<>(aggregator)) {
@@ -161,7 +163,7 @@ class TestDynamicWriteResultAggregator {
testHarness =
new OneInputStreamOperatorTestHarness<>(
new DynamicWriteResultAggregator(
- CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize))) {
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize,
"sinkId"))) {
testHarness.open();
TableKey tableKey = new TableKey("table", "branch");
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittable.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittable.java
index 4f0b68573f..1dec44e195 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittable.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittable.java
@@ -26,8 +26,10 @@ import
org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
/**
* The aggregated results of a single checkpoint which should be committed.
Containing the
- * serialized {@link DeltaManifests} files - which contains the commit data,
and the jobId,
- * operatorId, checkpointId triplet to identify the specific commit.
+ * serialized {@link DeltaManifests} files - which contains the commit data,
and the jobId, sinkId,
+ * operatorId, checkpointId to identify the specific commit.
+ *
+ * <p>Committables restored from older state use the jobId as sinkId.
*
* <p>{@link DynamicCommittableSerializer} is used to serialize {@link
DynamicCommittable} between
* the {@link DynamicWriter} and the {@link DynamicWriteResultAggregator}.
@@ -37,14 +39,21 @@ class DynamicCommittable implements Serializable {
private final TableKey key;
private final byte[][] manifests;
private final String jobId;
+ private final String sinkId;
private final String operatorId;
private final long checkpointId;
DynamicCommittable(
- TableKey key, byte[][] manifests, String jobId, String operatorId, long
checkpointId) {
+ TableKey key,
+ byte[][] manifests,
+ String jobId,
+ String sinkId,
+ String operatorId,
+ long checkpointId) {
this.key = key;
this.manifests = manifests;
this.jobId = jobId;
+ this.sinkId = sinkId;
this.operatorId = operatorId;
this.checkpointId = checkpointId;
}
@@ -61,6 +70,10 @@ class DynamicCommittable implements Serializable {
return jobId;
}
+ String sinkId() {
+ return sinkId;
+ }
+
String operatorId() {
return operatorId;
}
@@ -80,12 +93,14 @@ class DynamicCommittable implements Serializable {
&& Objects.equals(key, that.key)
&& Arrays.deepEquals(manifests, that.manifests)
&& Objects.equals(jobId, that.jobId)
+ && Objects.equals(sinkId, that.sinkId)
&& Objects.equals(operatorId, that.operatorId);
}
@Override
public int hashCode() {
- return Objects.hash(key, Arrays.deepHashCode(manifests), jobId,
operatorId, checkpointId);
+ return Objects.hash(
+ key, Arrays.deepHashCode(manifests), jobId, sinkId, operatorId,
checkpointId);
}
@Override
@@ -93,6 +108,7 @@ class DynamicCommittable implements Serializable {
return MoreObjects.toStringHelper(this)
.add("key", key)
.add("jobId", jobId)
+ .add("sinkId", sinkId)
.add("checkpointId", checkpointId)
.add("operatorId", operatorId)
.toString();
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java
index d599d29dba..e4f3808cd8 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java
@@ -33,10 +33,11 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
private static final int VERSION_1 = 1;
private static final int VERSION_2 = 2;
+ private static final int VERSION_3 = 3;
@Override
public int getVersion() {
- return VERSION_2;
+ return VERSION_3;
}
@Override
@@ -45,6 +46,7 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
committable.key().serializeTo(view);
view.writeUTF(committable.jobId());
+ view.writeUTF(committable.sinkId());
view.writeUTF(committable.operatorId());
view.writeLong(committable.checkpointId());
@@ -65,6 +67,8 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
return deserializeV1(serialized);
} else if (version == VERSION_2) {
return deserializeV2(serialized);
+ } else if (version == VERSION_3) {
+ return deserializeV3(serialized);
}
throw new IOException("Unrecognized version or corrupt state: " + version);
@@ -83,6 +87,7 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
new TableKey(key.tableName(), key.branch()),
new byte[][] {manifestBuf},
jobId,
+ jobId,
operatorId,
checkpointId);
}
@@ -93,7 +98,22 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
String jobId = view.readUTF();
String operatorId = view.readUTF();
long checkpointId = view.readLong();
+ byte[][] manifests = readManifests(view);
+ return new DynamicCommittable(key, manifests, jobId, jobId, operatorId,
checkpointId);
+ }
+
+ private DynamicCommittable deserializeV3(byte[] serialized) throws
IOException {
+ DataInputDeserializer view = new DataInputDeserializer(serialized);
+ TableKey key = TableKey.deserializeFrom(view);
+ String jobId = view.readUTF();
+ String sinkId = view.readUTF();
+ String operatorId = view.readUTF();
+ long checkpointId = view.readLong();
+ byte[][] manifests = readManifests(view);
+ return new DynamicCommittable(key, manifests, jobId, sinkId, operatorId,
checkpointId);
+ }
+ private static byte[][] readManifests(DataInputDeserializer view) throws
IOException {
byte[][] manifestsBuf = new byte[view.readInt()][];
for (int i = 0; i < manifestsBuf.length; i++) {
byte[] manifest = new byte[view.readInt()];
@@ -101,6 +121,6 @@ class DynamicCommittableSerializer implements
SimpleVersionedSerializer<DynamicC
manifestsBuf[i] = manifest;
}
- return new DynamicCommittable(key, manifestsBuf, jobId, operatorId,
checkpointId);
+ return manifestsBuf;
}
}
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
index c8b9a097ad..7a4bdf86aa 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
@@ -70,7 +70,7 @@ import org.slf4j.LoggerFactory;
* <li>There is no late checkpoint - if checkpoint 'x' has received in one
call, then after a
* successful run only checkpoints > x will arrive
* <li>There is no other writer which would generate another commit to the
same branch with the
- * same jobId-operatorId-checkpointId triplet
+ * same sinkId-operatorId-checkpointId triplet
* </ul>
*/
@Internal
@@ -82,6 +82,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
private static final long INITIAL_CHECKPOINT_ID = -1L;
private static final String FLINK_JOB_ID = "flink.job-id";
+ private static final String SINK_ID = "flink.sink-id";
private static final String OPERATOR_ID = "flink.operator-id";
private final Map<String, String> snapshotProperties;
private final boolean replacePartitions;
@@ -118,9 +119,8 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
| | | |
| | | +-- commit
requests at that checkpoint
| | +-- checkpointId, sorted
ascending so older commits go first
- | +-- (jobId, operatorId) of the producing
aggregator; deduplication against the
- | table's snapshot summaries is per (jobId,
operatorId), so each group is
- | walked against the ancestor chain
independently
+ | +-- (sinkId, operatorId) of the producing
aggregator; each group is
+ | deduplicated against the ancestor chain
independently
+-- (tableName, branch); we load the table and its ancestor
chain once per TableKey
Each (table, branch, checkpoint) triplet must have only one commit
request.
@@ -156,17 +156,17 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
List<Map.Entry<JobOperatorKey, NavigableMap<Long,
List<CommitRequest<DynamicCommittable>>>>>
jobEntries = Lists.newArrayList(tableEntry.getValue().entrySet());
- // Preserve checkpoint order across groups so that older-jobId commits
land before newer-jobId
- // ones when the batch mixes committables from different jobIds (e.g.
state replay after a
- // restart). Within a (jobId, operatorId) group, checkpoint order is
already guaranteed by
- // the inner NavigableMap.
+ // Preserve checkpoint order across groups so that older-sinkId commits
land before
+ // newer-sinkId ones when the batch mixes committables from different
runs (e.g. state replay
+ // after a restart). Within a (sinkId, operatorId) group, checkpoint
order is already
+ // guaranteed by the inner NavigableMap.
jobEntries.sort(Comparator.comparingLong(entry ->
entry.getValue().firstKey()));
for (Map.Entry<JobOperatorKey, NavigableMap<Long,
List<CommitRequest<DynamicCommittable>>>>
jobEntry : jobEntries) {
JobOperatorKey jobKey = jobEntry.getKey();
long maxCommittedCheckpointId =
- getMaxCommittedCheckpointId(ancestors, jobKey.jobId(),
jobKey.operatorId());
+ getMaxCommittedCheckpointId(ancestors, jobKey.sinkId(),
jobKey.operatorId());
NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>
skippedCommitRequests =
jobEntry.getValue().headMap(maxCommittedCheckpointId, true);
@@ -181,22 +181,24 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
jobEntry.getValue().tailMap(maxCommittedCheckpointId, false);
if (!uncommitted.isEmpty()) {
commitPendingRequests(
- table, tableKey.branch(), uncommitted, jobKey.jobId(),
jobKey.operatorId());
+ table,
+ tableKey.branch(),
+ uncommitted,
+ jobKey.jobId(),
+ jobKey.sinkId(),
+ jobKey.operatorId());
}
}
}
}
private static long getMaxCommittedCheckpointId(
- Iterable<Snapshot> ancestors, String flinkJobId, String operatorId) {
+ Iterable<Snapshot> ancestors, String sinkId, String operatorId) {
long lastCommittedCheckpointId = INITIAL_CHECKPOINT_ID;
for (Snapshot ancestor : ancestors) {
Map<String, String> summary = ancestor.summary();
- String snapshotFlinkJobId = summary.get(FLINK_JOB_ID);
- String snapshotOperatorId = summary.get(OPERATOR_ID);
- if (flinkJobId.equals(snapshotFlinkJobId)
- && (snapshotOperatorId == null ||
snapshotOperatorId.equals(operatorId))) {
+ if (writtenBySameSink(summary, sinkId, operatorId)) {
String value = summary.get(MAX_COMMITTED_CHECKPOINT_ID);
if (value != null) {
lastCommittedCheckpointId = Long.parseLong(value);
@@ -208,6 +210,18 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
return lastCommittedCheckpointId;
}
+ private static boolean writtenBySameSink(
+ Map<String, String> summary, String sinkId, String operatorId) {
+ String snapshotOperatorId = summary.get(OPERATOR_ID);
+ if (snapshotOperatorId != null && !snapshotOperatorId.equals(operatorId)) {
+ return false;
+ }
+
+ // Snapshots written before the sinkId was introduced are identified by
the jobId
+ String snapshotSinkId = summary.get(SINK_ID);
+ return sinkId.equals(snapshotSinkId != null ? snapshotSinkId :
summary.get(FLINK_JOB_ID));
+ }
+
/**
* Commits the data to the Iceberg table by reading the file data from the
{@link DeltaManifests}
* ordered by the checkpointId, and writing the new snapshot to the Iceberg
table. The {@link
@@ -216,6 +230,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
*
* @param commitRequestMap The checkpointId to {@link CommitRequest} map of
the changes to commit
* @param newFlinkJobId The jobId to store in the {@link SnapshotSummary}
+ * @param sinkId The sinkId to store in the {@link SnapshotSummary}
* @param operatorId The operatorId to store in the {@link SnapshotSummary}
* @throws IOException On commit failure
*/
@@ -224,6 +239,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
String branch,
NavigableMap<Long, List<CommitRequest<DynamicCommittable>>>
commitRequestMap,
String newFlinkJobId,
+ String sinkId,
String operatorId)
throws IOException {
long checkpointId = commitRequestMap.lastKey();
@@ -258,9 +274,9 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
}
if (replacePartitions) {
- replacePartitions(table, branch, pendingResults, newFlinkJobId,
operatorId);
+ replacePartitions(table, branch, pendingResults, newFlinkJobId, sinkId,
operatorId);
} else {
- commitDeltaTxn(table, branch, pendingResults, newFlinkJobId, operatorId);
+ commitDeltaTxn(table, branch, pendingResults, newFlinkJobId, sinkId,
operatorId);
}
FlinkManifestUtil.deleteCommittedManifests(table, manifests,
newFlinkJobId, checkpointId);
@@ -271,6 +287,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
String branch,
NavigableMap<Long, List<WriteResult>> pendingResults,
String newFlinkJobId,
+ String sinkId,
String operatorId) {
// Iceberg tables are unsorted. So the order of the append data does not
matter.
// Hence, we commit everything in one snapshot.
@@ -292,6 +309,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
summary,
"dynamic partition overwrite",
newFlinkJobId,
+ sinkId,
operatorId,
pendingResults.lastKey());
}
@@ -301,6 +319,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
String branch,
NavigableMap<Long, List<WriteResult>> pendingResults,
String newFlinkJobId,
+ String sinkId,
String operatorId) {
for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
long checkpointId = e.getKey();
@@ -328,7 +347,15 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
// pending checkpoints here are very rare to occur, i.e. only with very
short checkpoint
// intervals or when concurrent checkpointing is enabled.
commitOperation(
- table, branch, rowDelta, summary, "rowDelta", newFlinkJobId,
operatorId, checkpointId);
+ table,
+ branch,
+ rowDelta,
+ summary,
+ "rowDelta",
+ newFlinkJobId,
+ sinkId,
+ operatorId,
+ checkpointId);
}
}
@@ -340,20 +367,20 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
private static class MaxCommittedCheckpointIdValidator implements
SnapshotAncestryValidator {
private final long stagedCheckpointId;
- private final String flinkJobId;
+ private final String sinkId;
private final String flinkOperatorId;
private MaxCommittedCheckpointIdValidator(
- long stagedCheckpointId, String flinkJobId, String flinkOperatorId) {
+ long stagedCheckpointId, String sinkId, String flinkOperatorId) {
this.stagedCheckpointId = stagedCheckpointId;
- this.flinkJobId = flinkJobId;
+ this.sinkId = sinkId;
this.flinkOperatorId = flinkOperatorId;
}
@Override
public boolean validate(Iterable<Snapshot> baseSnapshots) {
long maxCommittedCheckpointId =
- getMaxCommittedCheckpointId(baseSnapshots, flinkJobId,
flinkOperatorId);
+ getMaxCommittedCheckpointId(baseSnapshots, sinkId, flinkOperatorId);
if (maxCommittedCheckpointId >= stagedCheckpointId) {
throw new MaxCommittedCheckpointMismatchException();
}
@@ -370,6 +397,7 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
CommitSummary summary,
String description,
String newFlinkJobId,
+ String sinkId,
String operatorId,
long checkpointId) {
@@ -385,10 +413,11 @@ class DynamicCommitter implements
Committer<DynamicCommittable> {
// used by the sink.
operation.set(MAX_COMMITTED_CHECKPOINT_ID, Long.toString(checkpointId));
operation.set(FLINK_JOB_ID, newFlinkJobId);
+ operation.set(SINK_ID, sinkId);
+
operation.set(OPERATOR_ID, operatorId);
operation.toBranch(branch);
- operation.validateWith(
- new MaxCommittedCheckpointIdValidator(checkpointId, newFlinkJobId,
operatorId));
+ operation.validateWith(new MaxCommittedCheckpointIdValidator(checkpointId,
sinkId, operatorId));
long startNano = System.nanoTime();
try {
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
index 3d478ecd6f..65a904bc77 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
@@ -99,7 +99,7 @@ public class DynamicIcebergSink
this.cacheMaximumSize = cacheMaximumSize;
// We generate a random UUID every time when a sink is created.
// This is used to separate files generated by different sinks writing the
same table.
- // Also used to generate the aggregator operator name
+ // Also used to generate the aggregator operator name, and to deduplicate
commits
this.sinkId = UUID.randomUUID().toString();
this.forwardWriteResults = forwardWriteResults;
}
@@ -168,7 +168,7 @@ public class DynamicIcebergSink
.transform(
prefixIfNotNull(uidPrefix, sinkId + " Pre Commit"),
typeInformation,
- new DynamicWriteResultAggregator(catalogLoader, cacheMaximumSize))
+ new DynamicWriteResultAggregator(catalogLoader, cacheMaximumSize,
sinkId))
.uid(prefixIfNotNull(uidPrefix, "-pre-commit-topology"));
}
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java
index 41f01c6475..7bdbf5b9b1 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultAggregator.java
@@ -52,7 +52,7 @@ import org.slf4j.LoggerFactory;
/**
* Operator which aggregates the individual {@link WriteResult} objects to a
single {@link
- * DynamicCommittable} per checkpoint (storing the serialized {@link
DeltaManifests}, jobId,
+ * DynamicCommittable} per checkpoint (storing the serialized {@link
DeltaManifests}, jobId, sinkId,
* operatorId, checkpointId)
*/
class DynamicWriteResultAggregator
@@ -64,6 +64,7 @@ class DynamicWriteResultAggregator
private final CatalogLoader catalogLoader;
private final int cacheMaximumSize;
+ private final String sinkId;
private long lastCheckpointId = CheckpointIDCounter.INITIAL_CHECKPOINT_ID -
1;
@@ -77,9 +78,10 @@ class DynamicWriteResultAggregator
private transient int attemptId;
private transient Catalog catalog;
- DynamicWriteResultAggregator(CatalogLoader catalogLoader, int
cacheMaximumSize) {
+ DynamicWriteResultAggregator(CatalogLoader catalogLoader, int
cacheMaximumSize, String sinkId) {
this.catalogLoader = catalogLoader;
this.cacheMaximumSize = cacheMaximumSize;
+ this.sinkId = sinkId;
}
@Override
@@ -129,6 +131,7 @@ class DynamicWriteResultAggregator
entries.getKey(),
writeToManifests(entries.getKey().tableName(),
entries.getValue(), checkpointId),
getContainingTask().getEnvironment().getJobID().toString(),
+ sinkId,
getRuntimeContext().getOperatorUniqueID(),
checkpointId),
checkpointId,
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java
index 463c09475d..097f31b7ba 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/JobOperatorKey.java
@@ -22,10 +22,12 @@ import java.util.Objects;
class JobOperatorKey {
private final String jobId;
+ private final String sinkId;
private final String operatorId;
JobOperatorKey(DynamicCommittable committable) {
this.jobId = committable.jobId();
+ this.sinkId = committable.sinkId();
this.operatorId = committable.operatorId();
}
@@ -33,6 +35,10 @@ class JobOperatorKey {
return jobId;
}
+ String sinkId() {
+ return sinkId;
+ }
+
String operatorId() {
return operatorId;
}
@@ -47,11 +53,13 @@ class JobOperatorKey {
return false;
}
- return Objects.equals(jobId, that.jobId) && Objects.equals(operatorId,
that.operatorId);
+ return Objects.equals(jobId, that.jobId)
+ && Objects.equals(sinkId, that.sinkId)
+ && Objects.equals(operatorId, that.operatorId);
}
@Override
public int hashCode() {
- return Objects.hash(jobId, operatorId);
+ return Objects.hash(jobId, sinkId, operatorId);
}
}
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommittableSerializer.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommittableSerializer.java
index 16890d1f63..55da85b2e1 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommittableSerializer.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommittableSerializer.java
@@ -24,6 +24,7 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Set;
+import java.util.UUID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.runtime.jobgraph.OperatorID;
@@ -36,22 +37,42 @@ class TestDynamicCommittableSerializer {
new TableKey("table", "branch"),
new byte[][] {{3, 4}, {5, 6}},
JobID.generate().toHexString(),
+ UUID.randomUUID().toString(),
new OperatorID().toHexString(),
5);
@Test
void testV1() throws IOException {
+ String jobId = JobID.generate().toHexString();
+ // Older formats have no sinkId, the jobId is used instead
var committable =
new DynamicCommittable(
new TableKey("table", "branch"),
new byte[][] {{3, 4}},
- JobID.generate().toHexString(),
+ jobId,
+ jobId,
new OperatorID().toHexString(),
5);
DynamicCommittableSerializer serializer = new
DynamicCommittableSerializer();
assertThat(serializer.deserialize(1,
serializeV1(committable))).isEqualTo(committable);
}
+ @Test
+ void testV2() throws IOException {
+ String jobId = JobID.generate().toHexString();
+ // Older formats have no sinkId, the jobId is used instead
+ var committable =
+ new DynamicCommittable(
+ new TableKey("table", "branch"),
+ new byte[][] {{3, 4}, {5, 6}},
+ jobId,
+ jobId,
+ new OperatorID().toHexString(),
+ 5);
+ DynamicCommittableSerializer serializer = new
DynamicCommittableSerializer();
+ assertThat(serializer.deserialize(2,
serializeV2(committable))).isEqualTo(committable);
+ }
+
@Test
void testLatestVersion() throws IOException {
DynamicCommittableSerializer serializer = new
DynamicCommittableSerializer();
@@ -67,6 +88,22 @@ class TestDynamicCommittableSerializer {
.isInstanceOf(IOException.class);
}
+ byte[] serializeV2(DynamicCommittable committable) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
+ committable.key().serializeTo(view);
+ view.writeUTF(committable.jobId());
+ view.writeUTF(committable.operatorId());
+ view.writeLong(committable.checkpointId());
+ view.writeInt(committable.manifests().length);
+ for (byte[] manifest : committable.manifests()) {
+ view.writeInt(manifest.length);
+ view.write(manifest);
+ }
+
+ return out.toByteArray();
+ }
+
byte[] serializeV1(DynamicCommittable committable) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(out);
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java
index aecad67c1a..0133879ff1 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicCommitter.java
@@ -27,6 +27,7 @@ import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Map;
+import java.util.UUID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.connector.sink2.Committer.CommitRequest;
import org.apache.flink.api.connector.sink2.mocks.MockCommitRequest;
@@ -67,6 +68,7 @@ import org.junit.jupiter.params.provider.ValueSource;
class TestDynamicCommitter {
static final String DB = "db";
+ static final String SINK_ID = "sinkId";
static final String TABLE1 = "table";
static final String TABLE2 = "table2";
@@ -170,7 +172,8 @@ class TestDynamicCommitter {
TableKey tableKey3 = new TableKey(TABLE2, "branch2");
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -188,15 +191,18 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey1, deltaManifests1, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey1, deltaManifests1, jobId, SINK_ID, operatorId,
checkpointId));
CommitRequest<DynamicCommittable> commitRequest2 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey2, deltaManifests2, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey2, deltaManifests2, jobId, SINK_ID, operatorId,
checkpointId));
CommitRequest<DynamicCommittable> commitRequest3 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey3, deltaManifests3, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey3, deltaManifests3, jobId, SINK_ID, operatorId,
checkpointId));
dynamicCommitter.commit(Sets.newHashSet(commitRequest1, commitRequest2,
commitRequest3));
@@ -210,6 +216,7 @@ class TestDynamicCommitter {
.put("added-records", "42")
.put("changed-partition-count", "1")
.put("flink.job-id", jobId)
+ .put("flink.sink-id", SINK_ID)
.put("flink.max-committed-checkpoint-id", "" + checkpointId)
.put("flink.operator-id", operatorId)
.put("total-data-files", "1")
@@ -227,6 +234,7 @@ class TestDynamicCommitter {
.put("added-records", "42")
.put("changed-partition-count", "1")
.put("flink.job-id", jobId)
+ .put("flink.sink-id", SINK_ID)
.put("flink.max-committed-checkpoint-id", "" + checkpointId)
.put("flink.operator-id", operatorId)
.put("total-data-files", "1")
@@ -247,6 +255,7 @@ class TestDynamicCommitter {
.put("added-records", "42")
.put("changed-partition-count", "1")
.put("flink.job-id", jobId)
+ .put("flink.sink-id", SINK_ID)
.put("flink.max-committed-checkpoint-id", "" + checkpointId)
.put("flink.operator-id", operatorId)
.put("total-data-files", "1")
@@ -280,7 +289,8 @@ class TestDynamicCommitter {
TableKey tableKey = new TableKey(TABLE1, "branch");
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -294,13 +304,15 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId));
dynamicCommitter.commit(Sets.newHashSet(commitRequest));
CommitRequest<DynamicCommittable> oldCommitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId - 1));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId - 1));
// Old commits requests shouldn't affect the result
dynamicCommitter.commit(Sets.newHashSet(oldCommitRequest));
@@ -347,7 +359,8 @@ class TestDynamicCommitter {
previousCommitterMetrics);
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -356,6 +369,7 @@ class TestDynamicCommitter {
// Operator id is stable across Flink job restarts, jobId is not.
final String operatorId = new OperatorID().toHexString();
final String previousJobIdStr = previousJobId.toHexString();
+ final String previousSinkId = UUID.randomUUID().toString();
final int previousCheckpointId = 10;
byte[][] previousManifests =
@@ -364,7 +378,12 @@ class TestDynamicCommitter {
DynamicCommittable previousCommittable =
new DynamicCommittable(
- tableKey, previousManifests, previousJobIdStr, operatorId,
previousCheckpointId);
+ tableKey,
+ previousManifests,
+ previousJobIdStr,
+ previousSinkId,
+ operatorId,
+ previousCheckpointId);
previousCommitter.commit(Sets.newHashSet(new
MockCommitRequest<>(previousCommittable)));
table.refresh();
@@ -382,6 +401,7 @@ class TestDynamicCommitter {
newCommitterMetrics);
final String newJobIdStr = newJobId.toHexString();
+ final String newSinkId = UUID.randomUUID().toString();
final int newCheckpointId = previousCheckpointId + 1;
byte[][] previousManifestsNew =
@@ -396,12 +416,13 @@ class TestDynamicCommitter {
tableKey,
previousManifestsNew,
previousJobIdStr,
+ previousSinkId,
operatorId,
previousCheckpointId));
CommitRequest<DynamicCommittable> newCommitRequest =
new MockCommitRequest<>(
new DynamicCommittable(
- tableKey, newManifests, newJobIdStr, operatorId,
newCheckpointId));
+ tableKey, newManifests, newJobIdStr, newSinkId, operatorId,
newCheckpointId));
newCommitter.commit(Sets.newHashSet(replayedPreviousCommitRequest,
newCommitRequest));
@@ -411,12 +432,14 @@ class TestDynamicCommitter {
Snapshot first = Iterables.get(table.snapshots(), 0);
assertThat(first.summary())
.containsEntry("flink.job-id", previousJobIdStr)
+ .containsEntry("flink.sink-id", previousSinkId)
.containsEntry("flink.max-committed-checkpoint-id",
String.valueOf(previousCheckpointId))
.containsEntry("flink.operator-id", operatorId);
Snapshot second = Iterables.get(table.snapshots(), 1);
assertThat(second.summary())
.containsEntry("flink.job-id", newJobIdStr)
+ .containsEntry("flink.sink-id", newSinkId)
.containsEntry("flink.max-committed-checkpoint-id",
String.valueOf(newCheckpointId))
.containsEntry("flink.operator-id", operatorId);
}
@@ -441,7 +464,8 @@ class TestDynamicCommitter {
committerMetrics);
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -460,10 +484,12 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> oldRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, oldManifests, oldJobId,
operatorId, oldCheckpointId));
+ new DynamicCommittable(
+ tableKey, oldManifests, oldJobId, oldJobId, operatorId,
oldCheckpointId));
CommitRequest<DynamicCommittable> newRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, newManifests, newJobId,
operatorId, newCheckpointId));
+ new DynamicCommittable(
+ tableKey, newManifests, newJobId, newJobId, operatorId,
newCheckpointId));
// Hand the requests in reversed order; the committer must still land them
in checkpointId
// order on the snapshot chain.
@@ -506,7 +532,8 @@ class TestDynamicCommitter {
committerMetrics);
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -530,7 +557,8 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId));
// Upgrade the table version
UpdateProperties updateApi = table1.updateProperties();
@@ -567,7 +595,8 @@ class TestDynamicCommitter {
TableKey tableKey = new TableKey(TABLE1, "branch");
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -581,7 +610,8 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId));
dynamicCommitter.commit(Sets.newHashSet(commitRequest));
@@ -618,7 +648,8 @@ class TestDynamicCommitter {
assertThat(table.snapshots()).isEmpty();
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -636,21 +667,24 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey1, deltaManifests1, jobId,
operatorId, checkpointId1));
+ new DynamicCommittable(
+ tableKey1, deltaManifests1, jobId, jobId, operatorId,
checkpointId1));
byte[][] deltaManifests2 =
aggregator.writeToManifests(tableKey1.tableName(),
WRITE_RESULT_BY_SPEC_2, checkpointId1);
CommitRequest<DynamicCommittable> commitRequest2 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey1, deltaManifests2, jobId,
operatorId, checkpointId1));
+ new DynamicCommittable(
+ tableKey1, deltaManifests2, jobId, jobId, operatorId,
checkpointId1));
byte[][] deltaManifests3 =
aggregator.writeToManifests(tableKey2.tableName(),
WRITE_RESULT_BY_SPEC_2, checkpointId2);
CommitRequest<DynamicCommittable> commitRequest3 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey2, deltaManifests3, jobId,
operatorId, checkpointId2));
+ new DynamicCommittable(
+ tableKey2, deltaManifests3, jobId, jobId, operatorId,
checkpointId2));
boolean overwriteMode = false;
int workerPoolSize = 1;
@@ -715,7 +749,8 @@ class TestDynamicCommitter {
assertThat(table.snapshots()).isEmpty();
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -741,15 +776,18 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests1, jobId,
operatorId, checkpointId1));
+ new DynamicCommittable(
+ tableKey, deltaManifests1, jobId, jobId, operatorId,
checkpointId1));
CommitRequest<DynamicCommittable> commitRequest2 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests2, jobId,
operatorId, checkpointId2));
+ new DynamicCommittable(
+ tableKey, deltaManifests2, jobId, jobId, operatorId,
checkpointId2));
CommitRequest<DynamicCommittable> commitRequest3 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests3, jobId,
operatorId, checkpointId3));
+ new DynamicCommittable(
+ tableKey, deltaManifests3, jobId, jobId, operatorId,
checkpointId3));
boolean overwriteMode = false;
int workerPoolSize = 1;
@@ -860,7 +898,8 @@ class TestDynamicCommitter {
assertThat(table.snapshots()).isEmpty();
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -875,14 +914,16 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifest1, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifest1, jobId, jobId, operatorId,
checkpointId));
byte[][] deltaManifest2 =
aggregator.writeToManifests(tableKey.tableName(),
WRITE_RESULT_BY_SPEC_2, checkpointId);
CommitRequest<DynamicCommittable> commitRequest2 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifest2, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifest2, jobId, jobId, operatorId,
checkpointId));
boolean overwriteMode = false;
int workerPoolSize = 1;
@@ -930,7 +971,8 @@ class TestDynamicCommitter {
TableKey tableKey = new TableKey(TABLE1, "branch");
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -944,7 +986,8 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, deltaManifests, jobId,
operatorId, checkpointId));
+ new DynamicCommittable(
+ tableKey, deltaManifests, jobId, jobId, operatorId,
checkpointId));
dynamicCommitter.commit(Sets.newHashSet(commitRequest));
@@ -954,7 +997,7 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> overwriteRequest =
new MockCommitRequest<>(
new DynamicCommittable(
- tableKey, overwriteManifests, jobId, operatorId, checkpointId
+ 1));
+ tableKey, overwriteManifests, jobId, jobId, operatorId,
checkpointId + 1));
dynamicCommitter.commit(Sets.newHashSet(overwriteRequest));
@@ -987,7 +1030,8 @@ class TestDynamicCommitter {
assertThat(table.snapshots()).isEmpty();
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
OneInputStreamOperatorTestHarness aggregatorHarness =
new OneInputStreamOperatorTestHarness(aggregator);
aggregatorHarness.open();
@@ -1003,7 +1047,7 @@ class TestDynamicCommitter {
CommitRequest<DynamicCommittable> commitRequest1 =
new MockCommitRequest<>(
- new DynamicCommittable(tableKey, manifests, jobId, operatorId,
checkpointId));
+ new DynamicCommittable(tableKey, manifests, jobId, jobId,
operatorId, checkpointId));
Collection<CommitRequest<DynamicCommittable>> commitRequests =
Sets.newHashSet(commitRequest1);
int workerPoolSize = 1;
@@ -1055,6 +1099,118 @@ class TestDynamicCommitter {
.build());
}
+ @Test
+ void testCommitAfterStatelessRestartWithSameJobId() throws Exception {
+ Table table = catalog.loadTable(TableIdentifier.of(TABLE1));
+ DynamicWriteResultAggregator aggregator = openAggregator();
+ TableKey tableKey = new TableKey(TABLE1, "branch");
+ String jobId = JobID.generate().toHexString();
+ String operatorId = new OperatorID().toHexString();
+ String newSinkId = UUID.randomUUID().toString();
+
+ createCommitter()
+ .commit(
+ Sets.newHashSet(
+ commitRequest(
+ aggregator,
+ tableKey,
+ WRITE_RESULT_BY_SPEC,
+ jobId,
+ UUID.randomUUID().toString(),
+ operatorId,
+ 10)));
+
+ createCommitter()
+ .commit(
+ Sets.newHashSet(
+ commitRequest(
+ aggregator,
+ tableKey,
+ WRITE_RESULT_BY_SPEC_2,
+ jobId,
+ newSinkId,
+ operatorId,
+ 1)));
+
+ table.refresh();
+ assertThat(table.snapshots()).hasSize(2);
+
assertThat(table.snapshot(table.refs().get("branch").snapshotId()).summary())
+ .containsEntry("flink.job-id", jobId)
+ .containsEntry("flink.sink-id", newSinkId)
+ .containsEntry("flink.max-committed-checkpoint-id", "1")
+ .containsEntry("total-records", "66");
+ }
+
+ @Test
+ void testSkipsAlreadyCommittedOlderCommittable() throws Exception {
+ Table table = catalog.loadTable(TableIdentifier.of(TABLE1));
+ DynamicWriteResultAggregator aggregator = openAggregator();
+ TableKey tableKey = new TableKey(TABLE1, "branch");
+ String jobId = JobID.generate().toHexString();
+ String operatorId = new OperatorID().toHexString();
+
+ // Snapshot written before the sinkId was introduced
+ table
+ .newAppend()
+ .appendFile(DATA_FILE)
+ .set("flink.job-id", jobId)
+ .set("flink.operator-id", operatorId)
+ .set("flink.max-committed-checkpoint-id", "5")
+ .toBranch("branch")
+ .commit();
+
+ // Committables from older state use the jobId as sinkId
+ createCommitter()
+ .commit(
+ Sets.newHashSet(
+ commitRequest(
+ aggregator, tableKey, WRITE_RESULT_BY_SPEC, jobId, jobId,
operatorId, 5),
+ commitRequest(
+ aggregator, tableKey, WRITE_RESULT_BY_SPEC_2, jobId,
jobId, operatorId, 6)));
+
+ table.refresh();
+ assertThat(table.snapshots()).hasSize(2);
+
assertThat(table.snapshot(table.refs().get("branch").snapshotId()).summary())
+ .containsEntry("flink.sink-id", jobId)
+ .containsEntry("flink.max-committed-checkpoint-id", "6")
+ .containsEntry("total-records", "66");
+ }
+
+ private DynamicWriteResultAggregator openAggregator() throws Exception {
+ DynamicWriteResultAggregator aggregator =
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, SINK_ID);
+ OneInputStreamOperatorTestHarness aggregatorHarness =
+ new OneInputStreamOperatorTestHarness(aggregator);
+ aggregatorHarness.open();
+ return aggregator;
+ }
+
+ private static DynamicCommitter createCommitter() {
+ return new DynamicCommitter(
+ CATALOG_EXTENSION.catalog(),
+ Maps.newHashMap(),
+ false,
+ 1,
+ "sinkId",
+ new DynamicCommitterMetrics(new UnregisteredMetricsGroup()));
+ }
+
+ private static CommitRequest<DynamicCommittable> commitRequest(
+ DynamicWriteResultAggregator aggregator,
+ TableKey tableKey,
+ Map<Integer, Collection<WriteResult>> writeResults,
+ String jobId,
+ String sinkId,
+ String operatorId,
+ long checkpointId)
+ throws IOException {
+ byte[][] manifests =
+ aggregator.writeToManifests(tableKey.tableName(), writeResults,
checkpointId);
+ return new MockCommitRequest<>(
+ new DynamicCommittable(tableKey, manifests, jobId, sinkId, operatorId,
checkpointId));
+ }
+
interface CommitHook extends Serializable {
default Collection<CommitRequest<DynamicCommittable>> beforeCommit(
Collection<CommitRequest<DynamicCommittable>> commitRequests) {
@@ -1153,11 +1309,20 @@ class TestDynamicCommitter {
CommitSummary summary,
String description,
String newFlinkJobId,
+ String sinkId,
String operatorId,
long checkpointId) {
commitHook.beforeCommitOperation();
super.commitOperation(
- table, branch, operation, summary, description, newFlinkJobId,
operatorId, checkpointId);
+ table,
+ branch,
+ operation,
+ summary,
+ description,
+ newFlinkJobId,
+ sinkId,
+ operatorId,
+ checkpointId);
commitHook.afterCommitOperation();
}
}
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 9e2f00bca0..780aca8eb8 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -309,6 +309,37 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
runTest(rows);
}
+ @Test
+ void testAllTablesShareTheSinkId() throws Exception {
+ List<String> tableNames = List.of("t1", "t2", "t3", "t4");
+ List<DynamicIcebergDataImpl> rows = Lists.newArrayList();
+ for (String tableName : tableNames) {
+ rows.add(
+ new DynamicIcebergDataImpl(
+ SimpleDataUtil.SCHEMA,
+ tableName,
+ SnapshotRef.MAIN_BRANCH,
+ PartitionSpec.unpartitioned()));
+ }
+
+ executeDynamicSink(rows, env, true, 2, null);
+
+ Set<String> sinkIds = Sets.newHashSet();
+ for (String tableName : tableNames) {
+ Map<String, String> summary =
+ CATALOG_EXTENSION
+ .catalog()
+ .loadTable(TableIdentifier.of(DATABASE, tableName))
+ .currentSnapshot()
+ .summary();
+ assertThat(summary).containsKey("flink.sink-id");
+
assertThat(summary.get("flink.sink-id")).isNotEqualTo(summary.get("flink.job-id"));
+ sinkIds.add(summary.get("flink.sink-id"));
+ }
+
+ assertThat(sinkIds).hasSize(1);
+ }
+
@Test
void testNoShuffleTopology() throws Exception {
DataStream<DynamicIcebergDataImpl> dataStream =
@@ -1889,11 +1920,13 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
.toBranch(original.key().branch())
.commit();
+ // Committables restored from older state use the jobId as sinkId
DynamicCommittable replayed =
new DynamicCommittable(
original.key(),
original.manifests(),
PREVIOUS_JOB_ID,
+ PREVIOUS_JOB_ID,
original.operatorId(),
replayedCheckpointId);
List<Committer.CommitRequest<DynamicCommittable>> enriched =
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java
index f68e8dae17..3786cea425 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriteResultAggregator.java
@@ -64,7 +64,8 @@ class TestDynamicWriteResultAggregator {
CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table2"), new
Schema());
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
cacheMaximumSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize, "sinkId");
try (OneInputStreamOperatorTestHarness<
CommittableMessage<DynamicWriteResult>,
CommittableMessage<DynamicCommittable>>
testHarness = new OneInputStreamOperatorTestHarness<>(aggregator)) {
@@ -107,7 +108,8 @@ class TestDynamicWriteResultAggregator {
// Disable caching of ManifestOutputFileFactory.
final int zeroCacheSize = 0;
DynamicWriteResultAggregator aggregator =
- new DynamicWriteResultAggregator(CATALOG_EXTENSION.catalogLoader(),
zeroCacheSize);
+ new DynamicWriteResultAggregator(
+ CATALOG_EXTENSION.catalogLoader(), zeroCacheSize, "sinkId");
try (OneInputStreamOperatorTestHarness<
CommittableMessage<DynamicWriteResult>,
CommittableMessage<DynamicCommittable>>
testHarness = new OneInputStreamOperatorTestHarness<>(aggregator)) {
@@ -161,7 +163,7 @@ class TestDynamicWriteResultAggregator {
testHarness =
new OneInputStreamOperatorTestHarness<>(
new DynamicWriteResultAggregator(
- CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize))) {
+ CATALOG_EXTENSION.catalogLoader(), cacheMaximumSize,
"sinkId"))) {
testHarness.open();
TableKey tableKey = new TableKey("table", "branch");
