This is an automated email from the ASF dual-hosted git repository.
huaxingao 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 c71e7dfca1 Flink: Backport: Close the Sink V2 aggregator's table
loader (#18496) (#18698)
c71e7dfca1 is described below
commit c71e7dfca17633ae2bce15c486278e8e07629cea
Author: Maximilian Michels <[email protected]>
AuthorDate: Fri Oct 30 00:39:01 2026 +0100
Flink: Backport: Close the Sink V2 aggregator's table loader (#18496)
(#18698)
---
.../iceberg/flink/sink/IcebergWriteAggregator.java | 9 ++++++++
.../iceberg/flink/sink/TestIcebergCommitter.java | 27 ++++++++++++++++++++++
.../iceberg/flink/sink/IcebergWriteAggregator.java | 9 ++++++++
.../iceberg/flink/sink/TestIcebergCommitter.java | 27 ++++++++++++++++++++++
4 files changed, 72 insertions(+)
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
index 7f21c1c372..ffd31662e8 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
@@ -75,6 +75,15 @@ class IcebergWriteAggregator extends
AbstractStreamOperator<CommittableMessage<I
() -> table, table.properties(), flinkJobId, operatorId,
subTaskId, attemptId);
}
+ @Override
+ public void close() throws Exception {
+ try {
+ super.close();
+ } finally {
+ tableLoader.close();
+ }
+ }
+
@Override
public void finish() throws IOException {
prepareSnapshotPreBarrier(Long.MAX_VALUE);
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
index 8fa8f8be8c..39b682069b 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
@@ -28,6 +28,7 @@ import static org.assertj.core.api.Assumptions.assumeThat;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import java.io.File;
import java.io.IOException;
@@ -81,9 +82,13 @@ import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TestBase;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.CatalogLoader;
import org.apache.iceberg.flink.SimpleDataUtil;
import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.io.FileWriterFactory;
import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -183,6 +188,28 @@ class TestIcebergCommitter extends TestBase {
tableLoader = TableLoader.fromHadoopTable(tableLocation);
}
+ @TestTemplate
+ void closesAggregatorCatalogOnShutdown() throws Exception {
+ try (HadoopCatalog catalog =
+ spy(new HadoopCatalog(new Configuration(),
temporaryFolder.toURI().toString()))) {
+ TableIdentifier identifier = TableIdentifier.of("test_table");
+ catalog.createTable(identifier, TestFixtures.SCHEMA);
+ CatalogLoader catalogLoader = mock(CatalogLoader.class);
+ doReturn(catalog).when(catalogLoader).loadCatalog();
+ TableLoader catalogTableLoader = TableLoader.fromCatalog(catalogLoader,
identifier);
+
+ try (OneInputStreamOperatorTestHarness<
+ CommittableMessage<WriteResult>,
CommittableMessage<IcebergCommittable>>
+ harness =
+ new OneInputStreamOperatorTestHarness<>(
+ new IcebergWriteAggregator(catalogTableLoader))) {
+ harness.open();
+ }
+
+ verify(catalog).close();
+ }
+ }
+
@TestTemplate
public void testCommitTxnWithoutDataFiles() throws Exception {
IcebergCommitter committer = getCommitter();
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
index 4a4a789bf9..17b2661724 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
@@ -89,6 +89,15 @@ class IcebergWriteAggregator extends
AbstractStreamOperator<CommittableMessage<I
() -> table, table.properties(), flinkJobId, operatorId,
subTaskId, attemptId);
}
+ @Override
+ public void close() throws Exception {
+ try {
+ super.close();
+ } finally {
+ tableLoader.close();
+ }
+ }
+
@Override
public void finish() throws IOException {
prepareSnapshotPreBarrier(lastCheckpointId + 1);
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
index a079b0ca2a..91d146b275 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
@@ -28,6 +28,7 @@ import static org.assertj.core.api.Assumptions.assumeThat;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import java.io.File;
import java.io.IOException;
@@ -81,9 +82,13 @@ import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TestBase;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.CatalogLoader;
import org.apache.iceberg.flink.SimpleDataUtil;
import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.io.FileWriterFactory;
import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -183,6 +188,28 @@ class TestIcebergCommitter extends TestBase {
tableLoader = TableLoader.fromHadoopTable(tableLocation);
}
+ @TestTemplate
+ void closesAggregatorCatalogOnShutdown() throws Exception {
+ try (HadoopCatalog catalog =
+ spy(new HadoopCatalog(new Configuration(),
temporaryFolder.toURI().toString()))) {
+ TableIdentifier identifier = TableIdentifier.of("test_table");
+ catalog.createTable(identifier, TestFixtures.SCHEMA);
+ CatalogLoader catalogLoader = mock(CatalogLoader.class);
+ doReturn(catalog).when(catalogLoader).loadCatalog();
+ TableLoader catalogTableLoader = TableLoader.fromCatalog(catalogLoader,
identifier);
+
+ try (OneInputStreamOperatorTestHarness<
+ CommittableMessage<WriteResult>,
CommittableMessage<IcebergCommittable>>
+ harness =
+ new OneInputStreamOperatorTestHarness<>(
+ new IcebergWriteAggregator(catalogTableLoader))) {
+ harness.open();
+ }
+
+ verify(catalog).close();
+ }
+ }
+
@TestTemplate
public void testCommitTxnWithoutDataFiles() throws Exception {
IcebergCommitter committer = getCommitter();
