stevenzwu commented on code in PR #6528:
URL: https://github.com/apache/iceberg/pull/6528#discussion_r1062821467
##########
flink/v1.16/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java:
##########
@@ -554,21 +579,178 @@ public void testMultipleJobsWriteSameTable() throws
Exception {
assertFlinkManifests(0);
SimpleDataUtil.assertTableRows(table, tableRows);
assertSnapshotSize(i + 1);
- assertMaxCommittedCheckpointId(jobId, checkpointId + 1);
+ assertMaxCommittedCheckpointId(jobId, operatorId, checkpointId + 1);
}
}
}
+ @Test
+ public void testMultipleSinksWriteSameTable() throws Exception {
+ long timestamp = 0;
+ List<RowData> tableRows = Lists.newArrayList();
+
+ JobID[] jobs = new JobID[] {new JobID(), new JobID(), new JobID()};
+ OperatorID[] operatorIds1 =
+ new OperatorID[] {new OperatorID(), new OperatorID(), new
OperatorID()};
+ OperatorID[] operatorIds2 =
+ new OperatorID[] {new OperatorID(), new OperatorID(), new
OperatorID()};
+ for (int i = 0; i < 20; i++) {
+ int jobIndex = i % 3;
+ int checkpointId = i / 3;
+ JobID jobId = jobs[jobIndex];
+ OperatorID operatorId1 = operatorIds1[jobIndex];
+ OperatorID operatorId2 = operatorIds2[jobIndex];
+ try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 =
createStreamSink(jobId);
+ OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 =
createStreamSink(jobId)) {
+ harness1.getStreamConfig().setOperatorID(operatorId1);
+ harness1.setup();
+ harness1.open();
+ harness2.getStreamConfig().setOperatorID(operatorId2);
+ harness2.setup();
+ harness2.open();
+
+ assertSnapshotSize(2 * i);
+ assertMaxCommittedCheckpointId(jobId, operatorId1, checkpointId == 0 ?
-1 : checkpointId);
+ assertMaxCommittedCheckpointId(jobId, operatorId2, checkpointId == 0 ?
-1 : checkpointId);
+
+ List<RowData> rows1 =
Lists.newArrayList(SimpleDataUtil.createRowData(i, "word-1-" + i));
+ tableRows.addAll(rows1);
+
+ DataFile dataFile1 = writeDataFile(String.format("data-1-%d", i),
rows1);
+ harness1.processElement(of(dataFile1), ++timestamp);
+ harness1.snapshot(checkpointId + 1, ++timestamp);
+
+ List<RowData> rows2 =
Lists.newArrayList(SimpleDataUtil.createRowData(i, "word-2-" + i));
+ tableRows.addAll(rows2);
+
+ DataFile dataFile2 = writeDataFile(String.format("data-2-%d", i),
rows2);
+ harness2.processElement(of(dataFile2), ++timestamp);
+ harness2.snapshot(checkpointId + 1, ++timestamp);
+
+ assertFlinkManifests(2);
+
+ harness1.notifyOfCompletedCheckpoint(checkpointId + 1);
+ harness2.notifyOfCompletedCheckpoint(checkpointId + 1);
+
+ assertFlinkManifests(0);
+ SimpleDataUtil.assertTableRows(table, tableRows);
+ assertSnapshotSize(2 * i + 2);
+ assertMaxCommittedCheckpointId(jobId, operatorId1, checkpointId + 1);
+ assertMaxCommittedCheckpointId(jobId, operatorId2, checkpointId + 1);
+ }
+ }
+ }
+
+ @Test
+ public void testMultipleSinksRecoveryFromValidSnapshot() throws Exception {
+ long checkpointId = 0;
+ long timestamp = 0;
+ List<RowData> expectedRows = Lists.newArrayList();
+ OperatorSubtaskState snapshot1;
+ OperatorSubtaskState snapshot2;
+
+ JobID jobId = new JobID();
+ OperatorID operatorId1 = new OperatorID();
+ OperatorID operatorId2 = new OperatorID();
+ try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 =
createStreamSink(jobId);
+ OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 =
createStreamSink(jobId)) {
+ harness1.getStreamConfig().setOperatorID(operatorId1);
+ harness1.setup();
+ harness1.open();
+ harness2.getStreamConfig().setOperatorID(operatorId2);
+ harness2.setup();
+ harness2.open();
+
+ assertSnapshotSize(0);
+ assertMaxCommittedCheckpointId(jobId, operatorId1, -1L);
+ assertMaxCommittedCheckpointId(jobId, operatorId2, -1L);
+
+ RowData row1 = SimpleDataUtil.createRowData(1, "hello1");
+ expectedRows.add(row1);
+ DataFile dataFile1 = writeDataFile("data-1-1", ImmutableList.of(row1));
+
+ harness1.processElement(of(dataFile1), ++timestamp);
+ snapshot1 = harness1.snapshot(++checkpointId, ++timestamp);
+
+ RowData row2 = SimpleDataUtil.createRowData(1, "hello2");
+ expectedRows.add(row2);
+ DataFile dataFile2 = writeDataFile("data-1-2", ImmutableList.of(row2));
+
+ harness2.processElement(of(dataFile2), ++timestamp);
+ snapshot2 = harness2.snapshot(checkpointId, ++timestamp);
+ assertFlinkManifests(2);
+
+ // Only notify one of the committers
+ harness1.notifyOfCompletedCheckpoint(checkpointId);
+ assertFlinkManifests(1);
+
+ // Only the first row is committed at this point
+ SimpleDataUtil.assertTableRows(table, ImmutableList.of(row1));
+ assertSnapshotSize(1);
+ assertMaxCommittedCheckpointId(jobId, operatorId1, checkpointId);
+ assertMaxCommittedCheckpointId(jobId, operatorId2, -1);
+ }
+
+ // Restore from the given snapshot
+ try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 =
createStreamSink(jobId);
+ OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 =
createStreamSink(jobId)) {
+ harness1.getStreamConfig().setOperatorID(operatorId1);
Review Comment:
this didn't test the case where operatorId/jobVertexId can change.
##########
flink/v1.16/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java:
##########
@@ -134,6 +136,7 @@ class IcebergFilesCommitter extends
AbstractStreamOperator<Void>
public void initializeState(StateInitializationContext context) throws
Exception {
super.initializeState(context);
this.flinkJobId =
getContainingTask().getEnvironment().getJobID().toString();
+ this.operatorUniqueId = getRuntimeContext().getOperatorUniqueID();
Review Comment:
This can be a problem. it gets the `jobVertexId` (a UUID like string). I
verified it by setting the `uidPrefix=iceberg-sink` in `TestFlinkIcebergSink`.
```
FlinkSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
...
.uidPrefix("iceberg-sink")
.append();
````
Then I can see the `operatorUniqueId` value as jobVertexId of
`0e4eaa3db14ae3b6752b25172fd05c50`. It can change when the job graph changed
(e.g. adding a stateless map operator). Then we won't be able to match the
operator id properly.
Ideally, we want to use the [operator uid from
Flink](https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/ops/upgrading/#datastream-api),
which can be set by users. See the code from `FlinkSink`.
```
if (uidPrefix != null) {
committerStream = committerStream.uid(uidPrefix + "-committer");
}
```
if uid is not provided by application, Flink automatically generates a UUID
by default. Setting uid for stateful operator (like Iceberg committer) is a
[best
practice](https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/ops/production_ready/#set-uuids-for-all-operators).
Flink uses uid to match state description. if uid changed, old state won't be
matched and restored. uid works perfectly for this purpose. But I am not sure
what is the best way to retrieve the `uid` in the `initializeState` method.
I can see the operator name from debugger, which is also set by `FlinkSink`
based on `uidPrefix`. It will be unique as `uidPrefix` should be unique.
Nonetheless, it is still not the ideal candidate/replacement for `uid` though.
<img width="630" alt="image"
src="https://user-images.githubusercontent.com/1545663/210861950-3bdf1442-63d3-42ad-b6d5-88d5bd4c3bd1.png";>
--
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]
![https://user-images.githubusercontent.com/1545663/210861950-3bdf1442-63d3-42ad-b6d5-88d5bd4c3bd1.png"](https://user-images.githubusercontent.com/1545663/210861950-3bdf1442-63d3-42ad-b6d5-88d5bd4c3bd1.png"){kind=link}