Skip to content

Commit b149442

Browse files
manuzhangmanuzhang
committed
Fix Flink UT
1 parent 05a81c2 commit b149442

6 files changed

Lines changed: 63 additions & 39 deletions

File tree

‎flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1029,7 +1029,7 @@ public void testFlinkManifests() throws Exception {
10291029
// 2. Read the data files from manifests and assert.
10301030
List<DataFile> dataFiles =
10311031
FlinkManifestUtil.readDataFiles(
1032-
createTestingManifestFile(manifestPath), table.io(), table.specs());
1032+
createTestingManifestFile(manifestPath, dataFile1), table.io(), table.specs());
10331033
assertThat(dataFiles).hasSize(1);
10341034
TestHelpers.assertEquals(dataFile1, dataFiles.get(0));
10351035

@@ -1165,13 +1165,17 @@ public void testDeleteFiles() throws Exception {
11651165
}
11661166
}
11671167

1168-
private ManifestFile createTestingManifestFile(Path manifestPath) throws IOException {
1169-
try (ManifestWriter<DataFile> writer =
1168+
private ManifestFile createTestingManifestFile(Path manifestPath, DataFile dataFile)
1169+
throws IOException {
1170+
ManifestWriter<DataFile> writer =
11701171
ManifestFiles.write(
1172+
formatVersion,
11711173
PartitionSpec.unpartitioned(),
1172-
org.apache.iceberg.Files.localOutput(manifestPath.toFile()))) {
1173-
return writer.toManifestFile();
1174-
}
1174+
table.io().newOutputFile(manifestPath.toString()),
1175+
0L);
1176+
writer.add(dataFile);
1177+
writer.close();
1178+
return writer.toManifestFile();
11751179
}
11761180

11771181
private IcebergWriteAggregator buildIcebergWriteAggregator(String myJobId, String operatorId) {

‎flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -763,7 +763,7 @@ public void testFlinkManifests() throws Exception {
763763
// 2. Read the data files from manifests and assert.
764764
List<DataFile> dataFiles =
765765
FlinkManifestUtil.readDataFiles(
766-
createTestingManifestFile(manifestPath), table.io(), table.specs());
766+
createTestingManifestFile(manifestPath, dataFile1), table.io(), table.specs());
767767
assertThat(dataFiles).hasSize(1);
768768
TestHelpers.assertEquals(dataFile1, dataFiles.get(0));
769769

@@ -813,7 +813,7 @@ public void testDeleteFiles() throws Exception {
813813
// 2. Read the data files from manifests and assert.
814814
List<DataFile> dataFiles =
815815
FlinkManifestUtil.readDataFiles(
816-
createTestingManifestFile(manifestPath), table.io(), table.specs());
816+
createTestingManifestFile(manifestPath, dataFile1), table.io(), table.specs());
817817
assertThat(dataFiles).hasSize(1);
818818
TestHelpers.assertEquals(dataFile1, dataFiles.get(0));
819819

@@ -1119,13 +1119,17 @@ private FileAppenderFactory<RowData> createDeletableAppenderFactory() {
11191119
null);
11201120
}
11211121

1122-
private ManifestFile createTestingManifestFile(Path manifestPath) throws IOException {
1123-
try (ManifestWriter<DataFile> writer =
1122+
private ManifestFile createTestingManifestFile(Path manifestPath, DataFile dataFile)
1123+
throws IOException {
1124+
ManifestWriter<DataFile> writer =
11241125
ManifestFiles.write(
1126+
formatVersion,
11251127
PartitionSpec.unpartitioned(),
1126-
org.apache.iceberg.Files.localOutput(manifestPath.toFile()))) {
1127-
return writer.toManifestFile();
1128-
}
1128+
table.io().newOutputFile(manifestPath.toString()),
1129+
0L);
1130+
writer.add(dataFile);
1131+
writer.close();
1132+
return writer.toManifestFile();
11291133
}
11301134

11311135
private List<Path> assertFlinkManifests(int expectedCount) throws IOException {

‎flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1028,7 +1028,7 @@ public void testFlinkManifests() throws Exception {
10281028
// 2. Read the data files from manifests and assert.
10291029
List<DataFile> dataFiles =
10301030
FlinkManifestUtil.readDataFiles(
1031-
createTestingManifestFile(manifestPath), table.io(), table.specs());
1031+
createTestingManifestFile(manifestPath, dataFile1), table.io(), table.specs());
10321032
assertThat(dataFiles).hasSize(1);
10331033
TestHelpers.assertEquals(dataFile1, dataFiles.get(0));
10341034

@@ -1164,13 +1164,17 @@ public void testDeleteFiles() throws Exception {
11641164
}
11651165
}
11661166

1167-
private ManifestFile createTestingManifestFile(Path manifestPath) throws IOException {
1168-
try (ManifestWriter<DataFile> writer =
1167+
private ManifestFile createTestingManifestFile(Path manifestPath, DataFile dataFile)
1168+
throws IOException {
1169+
ManifestWriter<DataFile> writer =
11691170
ManifestFiles.write(
1171+
formatVersion,
11701172
PartitionSpec.unpartitioned(),
1171-
org.apache.iceberg.Files.localOutput(manifestPath.toFile()))) {
1172-
return writer.toManifestFile();
1173-
}
1173+
table.io().newOutputFile(manifestPath.toString()),
1174+
0L);
1175+
writer.add(dataFile);
1176+
writer.close();
1177+
return writer.toManifestFile();
11741178
}
11751179

11761180
private IcebergWriteAggregator buildIcebergWriteAggregator(String myJobId, String operatorId) {

‎flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -763,7 +763,7 @@ public void testFlinkManifests() throws Exception {
763763
// 2. Read the data files from manifests and assert.
764764
List<DataFile> dataFiles =
765765
FlinkManifestUtil.readDataFiles(
766-
createTestingManifestFile(manifestPath), table.io(), table.specs());
766+
createTestingManifestFile(manifestPath, dataFile1), table.io(), table.specs());
767767
assertThat(dataFiles).hasSize(1);
768768
TestHelpers.assertEquals(dataFile1, dataFiles.get(0));
769769

@@ -813,7 +813,7 @@ public void testDeleteFiles() throws Exception {
813813
// 2. Read the data files from manifests and assert.
814814
List<DataFile> dataFiles =
815815
FlinkManifestUtil.readDataFiles(
816-
createTestingManifestFile(manifestPath), table.io(), table.specs());
816+
createTestingManifestFile(manifestPath, dataFile1), table.io(), table.specs());
817817
assertThat(dataFiles).hasSize(1);
818818
TestHelpers.assertEquals(dataFile1, dataFiles.get(0));
819819

@@ -1119,13 +1119,17 @@ private FileAppenderFactory<RowData> createDeletableAppenderFactory() {
11191119
null);
11201120
}
11211121

1122-
private ManifestFile createTestingManifestFile(Path manifestPath) throws IOException {
1123-
try (ManifestWriter<DataFile> writer =
1122+
private ManifestFile createTestingManifestFile(Path manifestPath, DataFile dataFile)
1123+
throws IOException {
1124+
ManifestWriter<DataFile> writer =
11241125
ManifestFiles.write(
1126+
formatVersion,
11251127
PartitionSpec.unpartitioned(),
1126-
org.apache.iceberg.Files.localOutput(manifestPath.toFile()))) {
1127-
return writer.toManifestFile();
1128-
}
1128+
table.io().newOutputFile(manifestPath.toString()),
1129+
0L);
1130+
writer.add(dataFile);
1131+
writer.close();
1132+
return writer.toManifestFile();
11291133
}
11301134

11311135
private List<Path> assertFlinkManifests(int expectedCount) throws IOException {

‎flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1022,7 +1022,7 @@ public void testFlinkManifests() throws Exception {
10221022
// 2. Read the data files from manifests and assert.
10231023
List<DataFile> dataFiles =
10241024
FlinkManifestUtil.readDataFiles(
1025-
createTestingManifestFile(manifestPath), table.io(), table.specs());
1025+
createTestingManifestFile(manifestPath, dataFile1), table.io(), table.specs());
10261026
assertThat(dataFiles).hasSize(1);
10271027
TestHelpers.assertEquals(dataFile1, dataFiles.get(0));
10281028

@@ -1154,13 +1154,17 @@ public void testDeleteFiles() throws Exception {
11541154
}
11551155
}
11561156

1157-
private ManifestFile createTestingManifestFile(Path manifestPath) throws IOException {
1158-
try (ManifestWriter<DataFile> writer =
1157+
private ManifestFile createTestingManifestFile(Path manifestPath, DataFile dataFile)
1158+
throws IOException {
1159+
ManifestWriter<DataFile> writer =
11591160
ManifestFiles.write(
1161+
formatVersion,
11601162
PartitionSpec.unpartitioned(),
1161-
org.apache.iceberg.Files.localOutput(manifestPath.toFile()))) {
1162-
return writer.toManifestFile();
1163-
}
1163+
table.io().newOutputFile(manifestPath.toString()),
1164+
0L);
1165+
writer.add(dataFile);
1166+
writer.close();
1167+
return writer.toManifestFile();
11641168
}
11651169

11661170
private IcebergWriteAggregator buildIcebergWriteAggregator(String myJobId, String operatorId) {

‎flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -763,7 +763,7 @@ public void testFlinkManifests() throws Exception {
763763
// 2. Read the data files from manifests and assert.
764764
List<DataFile> dataFiles =
765765
FlinkManifestUtil.readDataFiles(
766-
createTestingManifestFile(manifestPath), table.io(), table.specs());
766+
createTestingManifestFile(manifestPath, dataFile1), table.io(), table.specs());
767767
assertThat(dataFiles).hasSize(1);
768768
TestHelpers.assertEquals(dataFile1, dataFiles.get(0));
769769

@@ -813,7 +813,7 @@ public void testDeleteFiles() throws Exception {
813813
// 2. Read the data files from manifests and assert.
814814
List<DataFile> dataFiles =
815815
FlinkManifestUtil.readDataFiles(
816-
createTestingManifestFile(manifestPath), table.io(), table.specs());
816+
createTestingManifestFile(manifestPath, dataFile1), table.io(), table.specs());
817817
assertThat(dataFiles).hasSize(1);
818818
TestHelpers.assertEquals(dataFile1, dataFiles.get(0));
819819

@@ -1119,13 +1119,17 @@ private FileAppenderFactory<RowData> createDeletableAppenderFactory() {
11191119
null);
11201120
}
11211121

1122-
private ManifestFile createTestingManifestFile(Path manifestPath) throws IOException {
1123-
try (ManifestWriter<DataFile> writer =
1122+
private ManifestFile createTestingManifestFile(Path manifestPath, DataFile dataFile)
1123+
throws IOException {
1124+
ManifestWriter<DataFile> writer =
11241125
ManifestFiles.write(
1126+
formatVersion,
11251127
PartitionSpec.unpartitioned(),
1126-
org.apache.iceberg.Files.localOutput(manifestPath.toFile()))) {
1127-
return writer.toManifestFile();
1128-
}
1128+
table.io().newOutputFile(manifestPath.toString()),
1129+
0L);
1130+
writer.add(dataFile);
1131+
writer.close();
1132+
return writer.toManifestFile();
11291133
}
11301134

11311135
private List<Path> assertFlinkManifests(int expectedCount) throws IOException {

0 commit comments

Comments
 (0)