Skip to content

Flink: Fix flink manifest location collision when there are multiple committers for multiple sink tables #3986

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rdblue merged 9 commits into from
Feb 4, 2022
Merged

Flink: Fix flink manifest location collision when there are multiple committers for multiple sink tables #3986

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
171d161
Flink: Fix manifest location conflict when exist same subtask
coolderli Jan 26, 2022
ad42b66
update
coolderli Jan 26, 2022
8ebb870
remove duplicate code on flink1.12 and flink1.13
coolderli Jan 27, 2022
dc8cd20
remove static
coolderli Jan 27, 2022
0b47dbf
use operatorId to generate manifest path
coolderli Jan 27, 2022
6b885a3
remove unused import in TestFlinkManifest
coolderli Jan 27, 2022
a823c00
update format order
coolderli Jan 28, 2022
35c66d9
update params order
coolderli Jan 28, 2022
67ae8fd
fix checkstyle in FlinkManifestUtil
coolderli Jan 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkManifestUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,10 +61,11 @@ static List<DataFile> readDataFiles(ManifestFile manifestFile, FileIO io) throws
}
}

static ManifestOutputFileFactory createOutputFileFactory(Table table, String flinkJobId, int subTaskId,
long attemptNumber) {
static ManifestOutputFileFactory createOutputFileFactory(Table table, String flinkJobId, String operatorUniqueId,
int subTaskId, long attemptNumber) {
TableOperations ops = ((HasTableOperations) table).operations();
return new ManifestOutputFileFactory(ops, table.io(), table.properties(), flinkJobId, subTaskId, attemptNumber);
return new ManifestOutputFileFactory(ops, table.io(), table.properties(), flinkJobId, operatorUniqueId,
subTaskId, attemptNumber);
}

static DeltaManifests writeCompletedFiles(WriteResult result,
Expand Down
4 changes: 3 additions & 1 deletion flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,9 @@ public void initializeState(StateInitializationContext context) throws Exception

int subTaskId = getRuntimeContext().getIndexOfThisSubtask();
int attemptId = getRuntimeContext().getAttemptNumber();
this.manifestOutputFileFactory = FlinkManifestUtil.createOutputFileFactory(table, flinkJobId, subTaskId, attemptId);
String operatorUniqueId = getRuntimeContext().getOperatorUniqueID();
this.manifestOutputFileFactory = FlinkManifestUtil.createOutputFileFactory(table, flinkJobId, operatorUniqueId,
subTaskId, attemptId);
this.maxCommittedCheckpointId = INITIAL_CHECKPOINT_ID;

this.checkpointsState = context.getOperatorStateStore().getListState(STATE_DESCRIPTOR);
Expand Down
8 changes: 5 additions & 3 deletions flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/ManifestOutputFileFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,23 +35,25 @@ class ManifestOutputFileFactory {
private final FileIO io;
private final Map<String, String> props;
private final String flinkJobId;
private final String operatorUniqueId;
private final int subTaskId;
private final long attemptNumber;
private final AtomicInteger fileCount = new AtomicInteger(0);

ManifestOutputFileFactory(TableOperations ops, FileIO io, Map<String, String> props,
String flinkJobId, int subTaskId, long attemptNumber) {
String flinkJobId, String operatorUniqueId, int subTaskId, long attemptNumber) {
this.ops = ops;
this.io = io;
this.props = props;
this.flinkJobId = flinkJobId;
this.operatorUniqueId = operatorUniqueId;
this.subTaskId = subTaskId;
this.attemptNumber = attemptNumber;
}

private String generatePath(long checkpointId) {
return FileFormat.AVRO.addExtension(String.format("%s-%05d-%d-%d-%05d", flinkJobId, subTaskId,
attemptNumber, checkpointId, fileCount.incrementAndGet()));
return FileFormat.AVRO.addExtension(String.format("%s-%s-%05d-%d-%d-%05d", flinkJobId, operatorUniqueId,
subTaskId, attemptNumber, checkpointId, fileCount.incrementAndGet()));
}

OutputFile create(long checkpointId) {
Expand Down
18 changes: 14 additions & 4 deletions flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkManifest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,9 +87,10 @@ public void before() throws IOException {
@Test
public void testIO() throws IOException {
String flinkJobId = newFlinkJobId();
String operatorId = newOperatorUniqueId();
for (long checkpointId = 1; checkpointId <= 3; checkpointId++) {
ManifestOutputFileFactory factory =
FlinkManifestUtil.createOutputFileFactory(table, flinkJobId, 1, 1);
FlinkManifestUtil.createOutputFileFactory(table, flinkJobId, operatorId, 1, 1);
final long curCkpId = checkpointId;

List<DataFile> dataFiles = generateDataFiles(10);
Expand Down Expand Up @@ -122,11 +123,12 @@ public void testIO() throws IOException {
public void testUserProvidedManifestLocation() throws IOException {
long checkpointId = 1;
String flinkJobId = newFlinkJobId();
String operatorId = newOperatorUniqueId();
File userProvidedFolder = tempFolder.newFolder();
Map<String, String> props = ImmutableMap.of(FLINK_MANIFEST_LOCATION, userProvidedFolder.getAbsolutePath() + "///");
ManifestOutputFileFactory factory = new ManifestOutputFileFactory(
((HasTableOperations) table).operations(), table.io(), props,
flinkJobId, 1, 1);
flinkJobId, operatorId, 1, 1);

List<DataFile> dataFiles = generateDataFiles(5);
DeltaManifests deltaManifests = FlinkManifestUtil.writeCompletedFiles(
Expand Down Expand Up @@ -156,7 +158,9 @@ public void testUserProvidedManifestLocation() throws IOException {
public void testVersionedSerializer() throws IOException {
long checkpointId = 1;
String flinkJobId = newFlinkJobId();
ManifestOutputFileFactory factory = FlinkManifestUtil.createOutputFileFactory(table, flinkJobId, 1, 1);
String operatorId = newOperatorUniqueId();
ManifestOutputFileFactory factory = FlinkManifestUtil.createOutputFileFactory(table, flinkJobId, operatorId,
1, 1);

List<DataFile> dataFiles = generateDataFiles(10);
List<DeleteFile> eqDeleteFiles = generateEqDeleteFiles(10);
Expand Down Expand Up @@ -186,7 +190,9 @@ public void testCompatibility() throws IOException {
// The v2 deserializer should be able to deserialize the v1 binary.
long checkpointId = 1;
String flinkJobId = newFlinkJobId();
ManifestOutputFileFactory factory = FlinkManifestUtil.createOutputFileFactory(table, flinkJobId, 1, 1);
String operatorId = newOperatorUniqueId();
ManifestOutputFileFactory factory = FlinkManifestUtil.createOutputFileFactory(table, flinkJobId, operatorId,
1, 1);

List<DataFile> dataFiles = generateDataFiles(10);
ManifestFile manifest = FlinkManifestUtil.writeDataFiles(factory.create(checkpointId), table.spec(), dataFiles);
Expand Down Expand Up @@ -271,4 +277,8 @@ private List<DeleteFile> generatePosDeleteFiles(int fileNum) throws IOException
private static String newFlinkJobId() {
return UUID.randomUUID().toString();
}

private static String newOperatorUniqueId() {
return UUID.randomUUID().toString();
}
}
8 changes: 6 additions & 2 deletions flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -599,8 +599,10 @@ public void testFlinkManifests() throws Exception {
harness.snapshot(checkpoint, ++timestamp);
List<Path> manifestPaths = assertFlinkManifests(1);
Path manifestPath = manifestPaths.get(0);
String operatorId = harness.getOneInputOperator().getOperatorID().toString();
Assert.assertEquals("File name should have the expected pattern.",
String.format("%s-%05d-%d-%d-%05d.avro", jobId, 0, 0, checkpoint, 1), manifestPath.getFileName().toString());
String.format("%s-%s-%05d-%d-%d-%05d.avro", jobId, operatorId, 0, 0, checkpoint, 1),
manifestPath.getFileName().toString());

// 2. Read the data files from manifests and assert.
List<DataFile> dataFiles = FlinkManifestUtil.readDataFiles(createTestingManifestFile(manifestPath), table.io());
Expand Down Expand Up @@ -640,8 +642,10 @@ public void testDeleteFiles() throws Exception {
harness.snapshot(checkpoint, ++timestamp);
List<Path> manifestPaths = assertFlinkManifests(1);
Path manifestPath = manifestPaths.get(0);
String operatorId = harness.getOneInputOperator().getOperatorID().toString();
Assert.assertEquals("File name should have the expected pattern.",
String.format("%s-%05d-%d-%d-%05d.avro", jobId, 0, 0, checkpoint, 1), manifestPath.getFileName().toString());
String.format("%s-%s-%05d-%d-%d-%05d.avro", jobId, operatorId, 0, 0, checkpoint, 1),
manifestPath.getFileName().toString());

// 2. Read the data files from manifests and assert.
List<DataFile> dataFiles = FlinkManifestUtil.readDataFiles(createTestingManifestFile(manifestPath), table.io());
Expand Down