This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b05616e523 [flink] Release both catalogs FlinkGenericCatalog wraps
(#9422)
b05616e523 is described below
commit b05616e5231661ef5ba7fc6f3d6c2961f9036f4c
Author: ZIHAN DAI <[email protected]>
AuthorDate: Fri Aug 28 18:20:18 2026 +1000
[flink] Release both catalogs FlinkGenericCatalog wraps (#9422)
---
.../apache/paimon/flink/FlinkGenericCatalog.java | 36 +++++++-
.../paimon/flink/FlinkGenericCatalogCloseTest.java | 96 ++++++++++++++++++++++
2 files changed, 129 insertions(+), 3 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java
index 5f7a05d885..022f3b6216 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalog.java
@@ -19,6 +19,7 @@
package org.apache.paimon.flink;
import org.apache.paimon.flink.procedure.ProcedureUtil;
+import org.apache.paimon.utils.ExceptionUtils;
import org.apache.flink.table.catalog.AbstractCatalog;
import org.apache.flink.table.catalog.Catalog;
@@ -71,13 +72,42 @@ public class FlinkGenericCatalog extends AbstractCatalog {
@Override
public void open() throws CatalogException {
paimon.open();
- flink.open();
+ boolean opened = false;
+ try {
+ flink.open();
+ opened = true;
+ } finally {
+ if (!opened) {
+ // the caller will not close a catalog that failed to open
+ closeQuietly(paimon);
+ }
+ }
}
@Override
public void close() throws CatalogException {
- paimon.close();
- flink.close();
+ Throwable failure = null;
+ try {
+ paimon.close();
+ } catch (Throwable t) {
+ failure = t;
+ }
+ try {
+ flink.close();
+ } catch (Throwable t) {
+ failure = ExceptionUtils.firstOrSuppressed(t, failure);
+ }
+ if (null != failure) {
+ ExceptionUtils.rethrow(failure);
+ }
+ }
+
+ private static void closeQuietly(Catalog catalog) {
+ try {
+ catalog.close();
+ } catch (Throwable ignored) {
+ // the failure that is already on its way out is the one worth
reporting
+ }
}
@Override
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkGenericCatalogCloseTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkGenericCatalogCloseTest.java
new file mode 100644
index 0000000000..7ad845af55
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkGenericCatalogCloseTest.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests that {@link FlinkGenericCatalog} releases both catalogs it wraps. */
+class FlinkGenericCatalogCloseTest {
+
+ private FlinkCatalog paimonCatalog() {
+ FlinkCatalog result = mock(FlinkCatalog.class);
+ when(result.getName()).thenReturn("paimon");
+ when(result.getDefaultDatabase()).thenReturn("default");
+ return result;
+ }
+
+ /** The two catalogs are separate resources; one failing must not strand
the other. */
+ @Test
+ void testCloseClosesTheFlinkCatalogWhenPaimonFails() {
+ FlinkCatalog paimon = paimonCatalog();
+ Catalog flink = mock(Catalog.class);
+ CatalogException expected = new CatalogException("paimon is down");
+ doThrow(expected).when(paimon).close();
+
+ FlinkGenericCatalog catalog = new FlinkGenericCatalog(paimon, flink);
+ assertThatThrownBy(catalog::close).isSameAs(expected);
+
+ verify(flink).close();
+ }
+
+ /** Both failing must surface the first one, with the second attached
rather than dropped. */
+ @Test
+ void testCloseKeepsBothFailures() {
+ FlinkCatalog paimon = paimonCatalog();
+ Catalog flink = mock(Catalog.class);
+ CatalogException paimonFailure = new CatalogException("paimon is
down");
+ CatalogException flinkFailure = new CatalogException("flink is down");
+ doThrow(paimonFailure).when(paimon).close();
+ doThrow(flinkFailure).when(flink).close();
+
+ FlinkGenericCatalog catalog = new FlinkGenericCatalog(paimon, flink);
+ assertThatThrownBy(catalog::close).isSameAs(paimonFailure);
+
assertThat(paimonFailure.getSuppressed()).containsExactly(flinkFailure);
+ }
+
+ /** open() is the same pairing in reverse: a caller does not close what
failed to open. */
+ @Test
+ void testOpenClosesThePaimonCatalogWhenFlinkFails() {
+ FlinkCatalog paimon = paimonCatalog();
+ Catalog flink = mock(Catalog.class);
+ CatalogException expected = new CatalogException("flink is down");
+ doThrow(expected).when(flink).open();
+
+ FlinkGenericCatalog catalog = new FlinkGenericCatalog(paimon, flink);
+ assertThatThrownBy(catalog::open).isSameAs(expected);
+
+ verify(paimon).close();
+ }
+
+ /** Control: nothing fails, both are closed exactly once and nothing is
thrown. */
+ @Test
+ void testCloseClosesBoth() {
+ FlinkCatalog paimon = paimonCatalog();
+ Catalog flink = mock(Catalog.class);
+
+ new FlinkGenericCatalog(paimon, flink).close();
+
+ verify(paimon).close();
+ verify(flink).close();
+ }
+}
