lrsb commented on code in PR #16011:
URL: https://github.com/apache/iceberg/pull/16011#discussion_r3130390015
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -1434,6 +1447,59 @@ public void beforeCommitOperation() {
}
}
+ /**
+ * Seeds an ancestor snapshot under a synthetic previous jobId and injects a
replay committable
+ * tagged with that jobId into the main batch. The seed uses {@code
table.newAppend()} directly
+ * rather than a side committer so the real committable's manifest stays
intact.
+ */
+ static class ReplayPreviousJobIdCommittableHook implements CommitHook {
+ static final String PREVIOUS_JOB_ID = JobID.generate().toHexString();
+
+ private static final String MAX_COMMITTED_CHECKPOINT_ID =
"flink.max-committed-checkpoint-id";
+ private static final String FLINK_JOB_ID = "flink.job-id";
+ private static final String OPERATOR_ID = "flink.operator-id";
+
+ // Static to survive Flink operator serialization.
+ private static boolean hasTriggered = false;
+
+ private final DataFile seedDataFile;
+
+ ReplayPreviousJobIdCommittableHook(DataFile seedDataFile) {
+ this.seedDataFile = seedDataFile;
+ hasTriggered = false;
+ }
+
+ @Override
+ public void
beforeCommit(Collection<Committer.CommitRequest<DynamicCommittable>> requests) {
+ if (hasTriggered || requests.isEmpty()) {
+ return;
+ }
+
+ hasTriggered = true;
+ DynamicCommittable original =
requests.iterator().next().getCommittable();
+
+ Table table =
+
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.parse(original.key().tableName()));
+ table
+ .newAppend()
+ .appendFile(seedDataFile)
+ .set(FLINK_JOB_ID, PREVIOUS_JOB_ID)
+ .set(OPERATOR_ID, original.operatorId())
+ .set(MAX_COMMITTED_CHECKPOINT_ID,
Long.toString(original.checkpointId()))
+ .toBranch(original.key().branch())
+ .commit();
+
+ DynamicCommittable replayed =
+ new DynamicCommittable(
+ original.key(),
+ original.manifests(),
+ PREVIOUS_JOB_ID,
+ original.operatorId(),
+ original.checkpointId());
+ requests.add(new MockCommitRequest<>(replayed));
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
