(iceberg) branch main updated: Flink 2.3: Close the Sink V2 aggregator's table loader (#18496)

[pvary]

This is an automated email from the ASF dual-hosted git repository.

pvary 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 16bbd8c885 Flink 2.3: Close the Sink V2 aggregator's table loader 
(#18496)
16bbd8c885 is described below

commit 16bbd8c88567de49dccc7662bc33cdc3f9ceb30a
Author: openinx <open...@gmail.com>
AuthorDate: Tue Oct 20 07:50:54 2026 -0700

    Flink 2.3: Close the Sink V2 aggregator's table loader (#18496)
---
 .../iceberg/flink/sink/IcebergWriteAggregator.java |  9 ++++++++
 .../iceberg/flink/sink/TestIcebergCommitter.java   | 27 ++++++++++++++++++++++
 2 files changed, 36 insertions(+)

diff --git 
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
 
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
index 4a4a789bf9..17b2661724 100644
--- 
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java
+++ 
b/flink/v2.3/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.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
 
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
index a079b0ca2a..91d146b275 100644
--- 
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergCommitter.java
+++ 
b/flink/v2.3/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();