This is an automated email from the ASF dual-hosted git repository. suxiaogang223 pushed a commit to branch codex/fix-oceanbase-temporary-hikari-leak in repository https://gitbox.apache.org/repos/asf/doris.git
commit fe6da055997d12db2a9591ddc6eaab74c0efdda8 Author: Socrates <[email protected]> AuthorDate: Fri Jul 31 12:26:25 2026 +0800 [fix](fe) Close transient JDBC resources ### What problem does this PR solve? Issue Number: None Related PR: #66269 Problem Summary: Streaming table discovery created a JDBC client without closing its Hikari pool after metadata collection, including failure paths. SPI-backed catalog creation could also construct a connector and then fail validation before registration, leaving its resources open. Close the streaming client in a finally block and close an unregistered catalog when creation validation fails. Add unit tests for both failure paths. ### Release note Fix transient JDBC and connector resource leaks during failed initialization. ### Check List (For Author) - Test: Unit Test - ./run-fe-ut.sh --run org.apache.doris.datasource.jdbc.client.JdbcOceanBaseClientTest,org.apache.doris.job.util.StreamingJobUtilsTest,org.apache.doris.datasource.CatalogFactoryTest - Behavior changed: Yes. Temporary clients and failed catalog connectors are closed. - Does this need documentation: No --- .../apache/doris/datasource/CatalogFactory.java | 20 ++- .../apache/doris/job/util/StreamingJobUtils.java | 174 +++++++++++---------- .../doris/datasource/CatalogFactoryTest.java | 55 +++++++ .../doris/job/util/StreamingJobUtilsTest.java | 20 +++ 4 files changed, 176 insertions(+), 93 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java index a5afd90dfc5..37d2897e27d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java @@ -175,18 +175,22 @@ public class CatalogFactory { catalog.setDefaultPropsIfMissing(isReplay); if (!isReplay) { - catalog.checkWhenCreating(); - // This will check if the customized access controller can be created successfully. - // If failed, it will throw exception and the catalog will not be created. try { - catalog.initAccessController(true); - } catch (Throwable e) { - LOG.warn("Failed to init access controller", e); - throw new DdlException("Failed to init access controller: " + e.getMessage()); + catalog.checkWhenCreating(); + // This will check if the customized access controller can be created successfully. + // If failed, it will throw exception and the catalog will not be created. + try { + catalog.initAccessController(true); + } catch (Throwable e) { + LOG.warn("Failed to init access controller", e); + throw new DdlException("Failed to init access controller: " + e.getMessage()); + } + } catch (DdlException e) { + catalog.onClose(); + throw e; } } return catalog; } } - diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java b/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java index 1ef6bf18093..0877e43a698 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java @@ -379,98 +379,102 @@ public class StreamingJobUtils { } JdbcClient jdbcClient = getJdbcClient(sourceType, properties); - String database = getRemoteDbName(sourceType, properties); - List<String> tablesNameList = jdbcClient.getTablesNameList(database); - if (tablesNameList.isEmpty()) { - throw new JobException("No tables found in database " + database); - } - Map<String, String> tableCreateProperties = getTableCreateProperties(targetProperties); - - List<String> noPrimaryKeyTables = new ArrayList<>(); - for (String table : tablesNameList) { - if (!includeTablesList.isEmpty() && !includeTablesList.contains(table)) { - log.info("Skip table {} in database {} as it does not in include_tables {}", table, database, - includeTables); - continue; + try { + String database = getRemoteDbName(sourceType, properties); + List<String> tablesNameList = jdbcClient.getTablesNameList(database); + if (tablesNameList.isEmpty()) { + throw new JobException("No tables found in database " + database); } + Map<String, String> tableCreateProperties = getTableCreateProperties(targetProperties); + + List<String> noPrimaryKeyTables = new ArrayList<>(); + for (String table : tablesNameList) { + if (!includeTablesList.isEmpty() && !includeTablesList.contains(table)) { + log.info("Skip table {} in database {} as it does not in include_tables {}", table, database, + includeTables); + continue; + } - // if set include_tables, exclude_tables is ignored - if (includeTablesList.isEmpty() - && !excludeTablesList.isEmpty() && excludeTablesList.contains(table)) { - log.info("Skip table {} in database {} as it in exclude_tables {}", table, database, - excludeTables); - continue; - } + // if set include_tables, exclude_tables is ignored + if (includeTablesList.isEmpty() + && !excludeTablesList.isEmpty() && excludeTablesList.contains(table)) { + log.info("Skip table {} in database {} as it in exclude_tables {}", table, database, + excludeTables); + continue; + } - List<String> primaryKeys = jdbcClient.getPrimaryKeys(database, table); - List<Column> columns = getColumns(jdbcClient, database, table, primaryKeys); - if (primaryKeys.isEmpty()) { - noPrimaryKeyTables.add(table); - } + List<String> primaryKeys = jdbcClient.getPrimaryKeys(database, table); + List<Column> columns = getColumns(jdbcClient, database, table, primaryKeys); + if (primaryKeys.isEmpty()) { + noPrimaryKeyTables.add(table); + } - // Resolve target (Doris) table name; defaults to source table name if not configured - String targetTableName = properties.getOrDefault( - DataSourceConfigKeys.TABLE + "." + table + "." - + DataSourceConfigKeys.TABLE_TARGET_TABLE_SUFFIX, - table).trim(); - - // Validate and apply exclude_columns for this table - Set<String> excludeColumns = parseExcludeColumns(properties, table); - if (!excludeColumns.isEmpty()) { - validateExcludeColumns(excludeColumns, table, columns, primaryKeys); - columns = columns.stream() - .filter(col -> !excludeColumns.contains(col.getName())) - .collect(Collectors.toList()); - } + // Resolve target (Doris) table name; defaults to source table name if not configured + String targetTableName = properties.getOrDefault( + DataSourceConfigKeys.TABLE + "." + table + "." + + DataSourceConfigKeys.TABLE_TARGET_TABLE_SUFFIX, + table).trim(); + + // Validate and apply exclude_columns for this table + Set<String> excludeColumns = parseExcludeColumns(properties, table); + if (!excludeColumns.isEmpty()) { + validateExcludeColumns(excludeColumns, table, columns, primaryKeys); + columns = columns.stream() + .filter(col -> !excludeColumns.contains(col.getName())) + .collect(Collectors.toList()); + } - // Convert Column to ColumnDefinition - List<ColumnDefinition> columnDefinitions = columns.stream().map(col -> { - DataType dataType = DataType.fromCatalogType(col.getType()); - return new ColumnDefinition(col.getName(), dataType, col.isAllowNull(), col.getComment()); - }).collect(Collectors.toList()); - - // Create DistributionDescriptor - DistributionDescriptor distribution = new DistributionDescriptor( - true, // isHash - true, // isAutoBucket - FeConstants.default_bucket_num, - primaryKeys - ); - - // Create CreateTableInfo - CreateTableInfo createtblInfo = new CreateTableInfo( - true, // ifNotExists - false, // isExternal - false, // isTemp - InternalCatalog.INTERNAL_CATALOG_NAME, // ctlName - targetDb, // dbName - targetTableName, // tableName - columnDefinitions, // columns - ImmutableList.of(), // indexes - "olap", // engineName - KeysType.UNIQUE_KEYS, // keysType - primaryKeys, // keys - "", // comment - PartitionTableInfo.EMPTY, // partitionTableInfo - distribution, // distribution - ImmutableList.of(), // rollups - new HashMap<>(tableCreateProperties), // properties - ImmutableMap.of(), // extProperties - ImmutableList.of() // clusterKeyColumnNames - ); - CreateTableCommand createtblCmd = new CreateTableCommand(Optional.empty(), createtblInfo); - // Key: source (PG/MySQL) table name; Value: command that creates the Doris target table - createtblCmds.put(table, createtblCmd); - } - if (createtblCmds.isEmpty()) { - throw new JobException("Can not found match table in database " + database); - } + // Convert Column to ColumnDefinition + List<ColumnDefinition> columnDefinitions = columns.stream().map(col -> { + DataType dataType = DataType.fromCatalogType(col.getType()); + return new ColumnDefinition(col.getName(), dataType, col.isAllowNull(), col.getComment()); + }).collect(Collectors.toList()); + + // Create DistributionDescriptor + DistributionDescriptor distribution = new DistributionDescriptor( + true, // isHash + true, // isAutoBucket + FeConstants.default_bucket_num, + primaryKeys + ); + + // Create CreateTableInfo + CreateTableInfo createtblInfo = new CreateTableInfo( + true, // ifNotExists + false, // isExternal + false, // isTemp + InternalCatalog.INTERNAL_CATALOG_NAME, // ctlName + targetDb, // dbName + targetTableName, // tableName + columnDefinitions, // columns + ImmutableList.of(), // indexes + "olap", // engineName + KeysType.UNIQUE_KEYS, // keysType + primaryKeys, // keys + "", // comment + PartitionTableInfo.EMPTY, // partitionTableInfo + distribution, // distribution + ImmutableList.of(), // rollups + new HashMap<>(tableCreateProperties), // properties + ImmutableMap.of(), // extProperties + ImmutableList.of() // clusterKeyColumnNames + ); + CreateTableCommand createtblCmd = new CreateTableCommand(Optional.empty(), createtblInfo); + // Key: source (PG/MySQL) table name; Value: command that creates the Doris target table + createtblCmds.put(table, createtblCmd); + } + if (createtblCmds.isEmpty()) { + throw new JobException("Can not found match table in database " + database); + } - if (!noPrimaryKeyTables.isEmpty()) { - throw new JobException("The following tables do not have primary key defined: " - + String.join(", ", noPrimaryKeyTables)); + if (!noPrimaryKeyTables.isEmpty()) { + throw new JobException("The following tables do not have primary key defined: " + + String.join(", ", noPrimaryKeyTables)); + } + return createtblCmds; + } finally { + jdbcClient.closeClient(); } - return createtblCmds; } public static List<Column> getColumns(JdbcClient jdbcClient, diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogFactoryTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogFactoryTest.java new file mode 100644 index 00000000000..400075343d1 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogFactoryTest.java @@ -0,0 +1,55 @@ +// 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.doris.datasource; + +import org.apache.doris.common.DdlException; +import org.apache.doris.connector.ConnectorFactory; +import org.apache.doris.connector.api.Connector; +import org.apache.doris.nereids.trees.plans.commands.CreateCatalogCommand; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.HashMap; +import java.util.Map; + +public class CatalogFactoryTest { + + @Test + public void testCloseConnectorWhenCreateValidationFails() throws Exception { + Map<String, String> properties = new HashMap<>(); + properties.put(CatalogMgr.CATALOG_TYPE_PROP, "jdbc"); + Connector connector = Mockito.mock(Connector.class); + Mockito.doThrow(new DdlException("validation failed")) + .when(connector).preCreateValidation(Mockito.any()); + CreateCatalogCommand command = new CreateCatalogCommand( + "jdbc_catalog", false, "", "", properties); + + try (MockedStatic<ConnectorFactory> factory = Mockito.mockStatic(ConnectorFactory.class)) { + factory.when(() -> ConnectorFactory.createConnector( + Mockito.eq("jdbc"), Mockito.anyMap(), Mockito.any())) + .thenReturn(connector); + + Assert.assertThrows(DdlException.class, () -> CatalogFactory.createFromCommand(1, command)); + + Mockito.verify(connector).close(); + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java index 93c19074cd1..b2e2b0a56be 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java @@ -23,12 +23,14 @@ import org.apache.doris.catalog.ScalarType; import org.apache.doris.datasource.jdbc.client.JdbcClient; import org.apache.doris.job.cdc.DataSourceConfigKeys; import org.apache.doris.job.common.DataSourceType; +import org.apache.doris.job.exception.JobException; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentMatchers; import org.mockito.Mock; +import org.mockito.MockedStatic; import org.mockito.Mockito; import org.mockito.MockitoAnnotations; @@ -250,4 +252,22 @@ public class StreamingJobUtilsTest { Assert.assertEquals("test_db", StreamingJobUtils.getRemoteDbName(DataSourceType.OCEANBASE, properties)); } + + @Test + public void testGenerateCreateTableCmdsClosesJdbcClientOnFailure() { + Map<String, String> properties = new HashMap<>(); + try (MockedStatic<StreamingJobUtils> utils = Mockito.mockStatic(StreamingJobUtils.class, + Mockito.CALLS_REAL_METHODS)) { + utils.when(() -> StreamingJobUtils.getJdbcClient(DataSourceType.OCEANBASE, properties)) + .thenReturn(jdbcClient); + utils.when(() -> StreamingJobUtils.getRemoteDbName(DataSourceType.OCEANBASE, properties)) + .thenReturn("test_db"); + Mockito.when(jdbcClient.getTablesNameList("test_db")).thenReturn(new ArrayList<>()); + + Assert.assertThrows(JobException.class, () -> StreamingJobUtils.generateCreateTableCmds( + "target_db", DataSourceType.OCEANBASE, properties, new HashMap<>())); + + Mockito.verify(jdbcClient).closeClient(); + } + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
