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 55794516bc Flink: Fix flaky TestFlinkIcebergSinkV2 > testDeleteStats()
(#18241)
55794516bc is described below
commit 55794516bcca63834a89018884fffc5a5b1bfdb0
Author: pvary <[email protected]>
AuthorDate: Mon Sep 28 11:37:57 2026 +0200
Flink: Fix flaky TestFlinkIcebergSinkV2 > testDeleteStats() (#18241)
---
.../org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java | 8 ++++++--
.../java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java | 7 +++++--
.../org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java | 8 ++++++--
.../java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java | 7 +++++--
.../org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java | 8 ++++++--
.../java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java | 7 +++++--
.../org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java | 8 ++++++--
.../java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java | 7 +++++--
8 files changed, 44 insertions(+), 16 deletions(-)
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
index fac1b8226d..1ecc8fec2a 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
@@ -35,6 +35,7 @@ import org.apache.iceberg.FileFormat;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotChanges;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.TableProperties;
@@ -46,6 +47,7 @@ import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.BeforeEach;
@@ -245,12 +247,14 @@ public class TestFlinkIcebergSinkV2 extends
TestFlinkIcebergSinkV2Base {
expectedRecords,
SnapshotRef.MAIN_BRANCH);
- SnapshotChanges changes = SnapshotChanges.builderFor(table).build();
+ // The committer may append an empty snapshot after the change log commit
+ Snapshot snapshot = Iterables.getOnlyElement(findValidSnapshots());
+ SnapshotChanges changes =
SnapshotChanges.builderFor(table).snapshot(snapshot).build();
DeleteFile deleteFile = changes.addedDeleteFiles().iterator().next();
String fromStat =
new String(
deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
DataFile dataFile = changes.addedDataFiles().iterator().next();
- assumeThat(fromStat).isEqualTo(dataFile.location());
+ assertThat(fromStat).isEqualTo(dataFile.location());
}
}
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
index 71cec082a0..83b4951b96 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
@@ -47,6 +47,7 @@ import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.BeforeEach;
@@ -228,13 +229,15 @@ public class TestIcebergSinkV2 extends
TestFlinkIcebergSinkV2Base {
expectedRecords,
SnapshotRef.MAIN_BRANCH);
- SnapshotChanges changes = SnapshotChanges.builderFor(table).build();
+ // The committer may append an empty snapshot after the change log commit
+ Snapshot snapshot = Iterables.getOnlyElement(findValidSnapshots());
+ SnapshotChanges changes =
SnapshotChanges.builderFor(table).snapshot(snapshot).build();
DeleteFile deleteFile = changes.addedDeleteFiles().iterator().next();
String fromStat =
new String(
deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
DataFile dataFile = changes.addedDataFiles().iterator().next();
- assumeThat(fromStat).isEqualTo(dataFile.location());
+ assertThat(fromStat).isEqualTo(dataFile.location());
}
protected void testChangeLogs(
diff --git
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
index fac1b8226d..1ecc8fec2a 100644
---
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
+++
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
@@ -35,6 +35,7 @@ import org.apache.iceberg.FileFormat;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotChanges;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.TableProperties;
@@ -46,6 +47,7 @@ import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.BeforeEach;
@@ -245,12 +247,14 @@ public class TestFlinkIcebergSinkV2 extends
TestFlinkIcebergSinkV2Base {
expectedRecords,
SnapshotRef.MAIN_BRANCH);
- SnapshotChanges changes = SnapshotChanges.builderFor(table).build();
+ // The committer may append an empty snapshot after the change log commit
+ Snapshot snapshot = Iterables.getOnlyElement(findValidSnapshots());
+ SnapshotChanges changes =
SnapshotChanges.builderFor(table).snapshot(snapshot).build();
DeleteFile deleteFile = changes.addedDeleteFiles().iterator().next();
String fromStat =
new String(
deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
DataFile dataFile = changes.addedDataFiles().iterator().next();
- assumeThat(fromStat).isEqualTo(dataFile.location());
+ assertThat(fromStat).isEqualTo(dataFile.location());
}
}
diff --git
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
index 71cec082a0..83b4951b96 100644
---
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
+++
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
@@ -47,6 +47,7 @@ import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.BeforeEach;
@@ -228,13 +229,15 @@ public class TestIcebergSinkV2 extends
TestFlinkIcebergSinkV2Base {
expectedRecords,
SnapshotRef.MAIN_BRANCH);
- SnapshotChanges changes = SnapshotChanges.builderFor(table).build();
+ // The committer may append an empty snapshot after the change log commit
+ Snapshot snapshot = Iterables.getOnlyElement(findValidSnapshots());
+ SnapshotChanges changes =
SnapshotChanges.builderFor(table).snapshot(snapshot).build();
DeleteFile deleteFile = changes.addedDeleteFiles().iterator().next();
String fromStat =
new String(
deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
DataFile dataFile = changes.addedDataFiles().iterator().next();
- assumeThat(fromStat).isEqualTo(dataFile.location());
+ assertThat(fromStat).isEqualTo(dataFile.location());
}
protected void testChangeLogs(
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
index fac1b8226d..1ecc8fec2a 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
@@ -35,6 +35,7 @@ import org.apache.iceberg.FileFormat;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotChanges;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.TableProperties;
@@ -46,6 +47,7 @@ import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.BeforeEach;
@@ -245,12 +247,14 @@ public class TestFlinkIcebergSinkV2 extends
TestFlinkIcebergSinkV2Base {
expectedRecords,
SnapshotRef.MAIN_BRANCH);
- SnapshotChanges changes = SnapshotChanges.builderFor(table).build();
+ // The committer may append an empty snapshot after the change log commit
+ Snapshot snapshot = Iterables.getOnlyElement(findValidSnapshots());
+ SnapshotChanges changes =
SnapshotChanges.builderFor(table).snapshot(snapshot).build();
DeleteFile deleteFile = changes.addedDeleteFiles().iterator().next();
String fromStat =
new String(
deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
DataFile dataFile = changes.addedDataFiles().iterator().next();
- assumeThat(fromStat).isEqualTo(dataFile.location());
+ assertThat(fromStat).isEqualTo(dataFile.location());
}
}
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
index 71cec082a0..83b4951b96 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
@@ -47,6 +47,7 @@ import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.BeforeEach;
@@ -228,13 +229,15 @@ public class TestIcebergSinkV2 extends
TestFlinkIcebergSinkV2Base {
expectedRecords,
SnapshotRef.MAIN_BRANCH);
- SnapshotChanges changes = SnapshotChanges.builderFor(table).build();
+ // The committer may append an empty snapshot after the change log commit
+ Snapshot snapshot = Iterables.getOnlyElement(findValidSnapshots());
+ SnapshotChanges changes =
SnapshotChanges.builderFor(table).snapshot(snapshot).build();
DeleteFile deleteFile = changes.addedDeleteFiles().iterator().next();
String fromStat =
new String(
deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
DataFile dataFile = changes.addedDataFiles().iterator().next();
- assumeThat(fromStat).isEqualTo(dataFile.location());
+ assertThat(fromStat).isEqualTo(dataFile.location());
}
protected void testChangeLogs(
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
index fac1b8226d..1ecc8fec2a 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkIcebergSinkV2.java
@@ -35,6 +35,7 @@ import org.apache.iceberg.FileFormat;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotChanges;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.TableProperties;
@@ -46,6 +47,7 @@ import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.BeforeEach;
@@ -245,12 +247,14 @@ public class TestFlinkIcebergSinkV2 extends
TestFlinkIcebergSinkV2Base {
expectedRecords,
SnapshotRef.MAIN_BRANCH);
- SnapshotChanges changes = SnapshotChanges.builderFor(table).build();
+ // The committer may append an empty snapshot after the change log commit
+ Snapshot snapshot = Iterables.getOnlyElement(findValidSnapshots());
+ SnapshotChanges changes =
SnapshotChanges.builderFor(table).snapshot(snapshot).build();
DeleteFile deleteFile = changes.addedDeleteFiles().iterator().next();
String fromStat =
new String(
deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
DataFile dataFile = changes.addedDataFiles().iterator().next();
- assumeThat(fromStat).isEqualTo(dataFile.location());
+ assertThat(fromStat).isEqualTo(dataFile.location());
}
}
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
index 71cec082a0..83b4951b96 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkV2.java
@@ -47,6 +47,7 @@ import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.BeforeEach;
@@ -228,13 +229,15 @@ public class TestIcebergSinkV2 extends
TestFlinkIcebergSinkV2Base {
expectedRecords,
SnapshotRef.MAIN_BRANCH);
- SnapshotChanges changes = SnapshotChanges.builderFor(table).build();
+ // The committer may append an empty snapshot after the change log commit
+ Snapshot snapshot = Iterables.getOnlyElement(findValidSnapshots());
+ SnapshotChanges changes =
SnapshotChanges.builderFor(table).snapshot(snapshot).build();
DeleteFile deleteFile = changes.addedDeleteFiles().iterator().next();
String fromStat =
new String(
deleteFile.lowerBounds().get(MetadataColumns.DELETE_FILE_PATH.fieldId()).array());
DataFile dataFile = changes.addedDataFiles().iterator().next();
- assumeThat(fromStat).isEqualTo(dataFile.location());
+ assertThat(fromStat).isEqualTo(dataFile.location());
}
protected void testChangeLogs(
