This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 6ddb9af9057 [fix](catalog) Make external catalog resource cleanup
ownership-safe (#66269)
6ddb9af9057 is described below
commit 6ddb9af90571d5ceb41d13caacd80ebbf753b913
Author: Socrates <[email protected]>
AuthorDate: Wed Aug 5 10:01:47 2026 +0800
[fix](catalog) Make external catalog resource cleanup ownership-safe
(#66269)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
External catalog creation can allocate resources before the catalog is
registered, including temporary JDBC clients, connection pools, and
access-controller plugins. Those temporary objects did not have a
consistent owner on validation failure or dry-run paths, so they could
outlive the catalog attempt.
This change makes cleanup ownership explicit:
- Separates unregistered-catalog creation-failure cleanup from
registered catalog teardown.
- Closes temporary JDBC clients after OceanBase compatibility detection
while preserving the returned normal-use client.
- Adds an AccessController lifecycle hook and closes dry-run controllers
and controllers removed with a catalog.
- Makes Ranger Hive controller shutdown safe: scheduled audit flushing
is cancelled, pending audit events are drained once, and plugin cleanup
waits for in-flight authorization.
- Corrects the Ranger audit flusher so each scheduled invocation
performs one flush instead of owning an infinite loop.
---
.../connector/jdbc/client/JdbcConnectorClient.java | 21 ++-
.../ranger/hive/RangerHiveAccessController.java | 110 +++++++++++--
.../ranger/hive/RangerHiveAuditHandler.java | 53 ++++++-
.../ranger/hive/RangerHiveAuditLogFlusher.java | 17 +-
.../apache/doris/datasource/CatalogFactory.java | 26 ++-
.../org/apache/doris/datasource/CatalogIf.java | 5 +
.../org/apache/doris/datasource/CatalogMgr.java | 90 ++++++-----
.../apache/doris/datasource/ExternalCatalog.java | 70 ++++++++-
.../jdbc/client/JdbcOceanBaseClient.java | 1 +
.../plugin/PluginDrivenExternalCatalog.java | 22 ++-
.../apache/doris/job/util/StreamingJobUtils.java | 174 +++++++++++----------
.../mysql/privilege/AccessControllerManager.java | 153 +++++++++++++++---
.../mysql/privilege/CatalogAccessController.java | 3 +
.../ranger/hive/RangerHiveAuditLogFlusherTest.java | 139 ++++++++++++++++
.../doris/datasource/CatalogFactoryTest.java | 90 +++++++++++
.../apache/doris/datasource/CatalogMgrTest.java | 60 +++++++
.../datasource/ExternalCatalogDeadlockTest.java | 17 +-
.../jdbc/client/JdbcOceanBaseClientTest.java | 105 +++++++++++++
.../doris/job/util/StreamingJobUtilsTest.java | 20 +++
.../privilege/AccessControllerManagerTest.java | 162 +++++++++++++++++++
20 files changed, 1133 insertions(+), 205 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-jdbc/src/main/java/org/apache/doris/connector/jdbc/client/JdbcConnectorClient.java
b/fe/fe-connector/fe-connector-jdbc/src/main/java/org/apache/doris/connector/jdbc/client/JdbcConnectorClient.java
index a3871e0005d..e1ad6c33b52 100644
---
a/fe/fe-connector/fe-connector-jdbc/src/main/java/org/apache/doris/connector/jdbc/client/JdbcConnectorClient.java
+++
b/fe/fe-connector/fe-connector-jdbc/src/main/java/org/apache/doris/connector/jdbc/client/JdbcConnectorClient.java
@@ -179,12 +179,21 @@ public abstract class JdbcConnectorClient implements
Closeable {
default:
throw new DorisConnectorException("Unsupported JDBC DB type: "
+ dbType);
}
- client.initializeClassLoader(driverUrl);
- String sanitizedUrl = urlSanitizer.apply(jdbcUrl);
- client.initializeDataSource(sanitizedUrl, user, password, driverClass,
- poolMinSize, poolMaxSize, poolMaxWaitTime, poolMaxLifeTime);
- client.postInitialize();
- return client;
+ try {
+ client.initializeClassLoader(driverUrl);
+ String sanitizedUrl = urlSanitizer.apply(jdbcUrl);
+ client.initializeDataSource(sanitizedUrl, user, password,
driverClass,
+ poolMinSize, poolMaxSize, poolMaxWaitTime,
poolMaxLifeTime);
+ client.postInitialize();
+ return client;
+ } catch (RuntimeException | Error e) {
+ try {
+ client.close();
+ } catch (RuntimeException | Error closeFailure) {
+ e.addSuppressed(closeFailure);
+ }
+ throw e;
+ }
}
protected JdbcConnectorClient(
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAccessController.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAccessController.java
index ae86e1ff14f..6f862ceade1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAccessController.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAccessController.java
@@ -24,7 +24,9 @@ import
org.apache.doris.catalog.authorizer.ranger.RangerAccessController;
import org.apache.doris.common.AuthorizationException;
import org.apache.doris.common.ThreadPoolManager;
import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.mysql.privilege.DataMaskPolicy;
import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.mysql.privilege.RowFilterPolicy;
import com.google.common.collect.Maps;
import org.apache.logging.log4j.LogManager;
@@ -42,17 +44,25 @@ import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
public class RangerHiveAccessController extends RangerAccessController {
private static final Logger LOG =
LogManager.getLogger(RangerHiveAccessController.class);
- private static ScheduledThreadPoolExecutor logFlushTimer =
ThreadPoolManager.newDaemonScheduledThreadPool(1,
+ private static final ScheduledThreadPoolExecutor LOG_FLUSH_TIMER =
ThreadPoolManager.newDaemonScheduledThreadPool(1,
"ranger-hive-audit-log-flusher-timer", true);
private RangerHivePlugin hivePlugin;
private RangerHiveAuditHandler auditHandler;
+ private ScheduledFuture<?> logFlushFuture;
+ // The manager can remove a controller while a query still holds its
reference. Keep the plugin alive
+ // until that query completes, then prevent any later authorization from
using the cleaned plugin.
+ private final ReentrantReadWriteLock lifecycleLock = new
ReentrantReadWriteLock();
+ private boolean closed;
public RangerHiveAccessController(Map<String, String> properties) {
this(properties, null);
@@ -64,7 +74,41 @@ public class RangerHiveAccessController extends
RangerAccessController {
hivePlugin = new RangerHivePlugin(serviceName,
rangerAuthContextListener);
auditHandler = new RangerHiveAuditHandler(hivePlugin.getConfig());
// start a timed log flusher
- logFlushTimer.scheduleAtFixedRate(new
RangerHiveAuditLogFlusher(auditHandler), 10, 20L, TimeUnit.SECONDS);
+ logFlushFuture = LOG_FLUSH_TIMER.scheduleAtFixedRate(
+ new RangerHiveAuditLogFlusher(auditHandler), 10, 20L,
TimeUnit.SECONDS);
+ }
+
+ @Override
+ public void close() {
+ lifecycleLock.writeLock().lock();
+ try {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ if (logFlushFuture != null) {
+ logFlushFuture.cancel(false);
+ logFlushFuture = null;
+ }
+ // flushAudit atomically drains the handler. This preserves events
produced before close without
+ // racing the periodic flusher or re-emitting events it has
already sent.
+ try {
+ auditHandler.flushAudit();
+ } catch (Throwable e) {
+ LOG.warn("Failed to flush Ranger Hive audit events while
closing the access controller", e);
+ }
+ if (hivePlugin != null) {
+ try {
+ hivePlugin.cleanup();
+ } catch (Throwable e) {
+ LOG.warn("Failed to clean up Ranger Hive plugin", e);
+ } finally {
+ hivePlugin = null;
+ }
+ }
+ } finally {
+ lifecycleLock.writeLock().unlock();
+ }
}
private RangerAccessRequestImpl createRequest(UserIdentity currentUser,
HiveAccessType accessType) {
@@ -95,24 +139,40 @@ public class RangerHiveAccessController extends
RangerAccessController {
private void checkPrivileges(UserIdentity currentUser, HiveAccessType
accessType,
List<RangerHiveResource> hiveResources) throws
AuthorizationException {
- List<RangerAccessRequest> requests = new ArrayList<>();
- for (RangerHiveResource resource : hiveResources) {
- RangerAccessRequestImpl request = createRequest(currentUser,
accessType);
- request.setResource(resource);
- requests.add(request);
- }
+ lifecycleLock.readLock().lock();
+ try {
+ if (closed) {
+ throw new AuthorizationException("Ranger Hive access
controller has been closed");
+ }
+ List<RangerAccessRequest> requests = new ArrayList<>();
+ for (RangerHiveResource resource : hiveResources) {
+ RangerAccessRequestImpl request = createRequest(currentUser,
accessType);
+ request.setResource(resource);
+ requests.add(request);
+ }
- Collection<RangerAccessResult> results =
hivePlugin.isAccessAllowed(requests, auditHandler);
- checkRequestResults(results, accessType.name());
+ Collection<RangerAccessResult> results =
hivePlugin.isAccessAllowed(requests, auditHandler);
+ checkRequestResults(results, accessType.name());
+ } finally {
+ lifecycleLock.readLock().unlock();
+ }
}
private boolean checkPrivilege(UserIdentity currentUser, HiveAccessType
accessType,
RangerHiveResource resource) {
- RangerAccessRequestImpl request = createRequest(currentUser,
accessType);
- request.setResource(resource);
+ lifecycleLock.readLock().lock();
+ try {
+ if (closed) {
+ return false;
+ }
+ RangerAccessRequestImpl request = createRequest(currentUser,
accessType);
+ request.setResource(resource);
- RangerAccessResult result = hivePlugin.isAccessAllowed(request,
auditHandler);
- return checkRequestResult(request, result, accessType.name());
+ RangerAccessResult result = hivePlugin.isAccessAllowed(request,
auditHandler);
+ return checkRequestResult(request, result, accessType.name());
+ } finally {
+ lifecycleLock.readLock().unlock();
+ }
}
private HiveAccessType convertToAccessType(PrivPredicate predicate) {
@@ -198,6 +258,28 @@ public class RangerHiveAccessController extends
RangerAccessController {
return true;
}
+ @Override
+ public List<? extends RowFilterPolicy> evalRowFilterPolicies(UserIdentity
currentUser, String ctl, String db,
+ String tbl) {
+ lifecycleLock.readLock().lock();
+ try {
+ return closed ? new ArrayList<>() :
super.evalRowFilterPolicies(currentUser, ctl, db, tbl);
+ } finally {
+ lifecycleLock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public Optional<DataMaskPolicy> evalDataMaskPolicy(UserIdentity
currentUser, String ctl, String db, String tbl,
+ String col) {
+ lifecycleLock.readLock().lock();
+ try {
+ return closed ? Optional.empty() :
super.evalDataMaskPolicy(currentUser, ctl, db, tbl, col);
+ } finally {
+ lifecycleLock.readLock().unlock();
+ }
+ }
+
@Override
protected RangerHiveResource createResource(String ctl, String db, String
tbl) {
return new RangerHiveResource(HiveObjectType.TABLE,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditHandler.java
index 73d73a3933b..276c1155cbe 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditHandler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditHandler.java
@@ -29,9 +29,11 @@ import
org.apache.ranger.plugin.policyengine.RangerAccessResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.Deque;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -66,7 +68,12 @@ public class RangerHiveAuditHandler extends
RangerDefaultAuditHandler {
private static final Logger LOG =
LoggerFactory.getLogger(RangerDefaultAuditHandler.class);
private final int requestQuerySize;
+ private final Object auditBufferLock = new Object();
+ private final Object flushLock = new Object();
private final Collection<AuthzAuditEvent> auditEvents = new ArrayList<>();
+ // Only accessed while holding flushLock. Successfully delivered events
are removed one by one, so a
+ // provider exception leaves the failed event and every later event
available for the next periodic tick.
+ private final Deque<AuthzAuditEvent> pendingAuditEvents = new
ArrayDeque<>();
private boolean deniedExists = false;
public RangerHiveAuditHandler() {
@@ -240,17 +247,43 @@ public class RangerHiveAuditHandler extends
RangerDefaultAuditHandler {
}
public void flushAudit() {
- for (AuthzAuditEvent auditEvent : auditEvents) {
- if (deniedExists && auditEvent.getAccessResult() != 0) { // if
deny exists, skip logging for allowed results
- continue;
+ synchronized (flushLock) {
+ Collection<AuthzAuditEvent> eventsToFlush;
+ boolean deniedExistsForEvents;
+ // Keep the producer critical section limited to snapshot/reset.
Provider delivery can block, but
+ // authorization callbacks remain free to enqueue into the next
batch.
+ synchronized (auditBufferLock) {
+ eventsToFlush = new ArrayList<>(auditEvents);
+ deniedExistsForEvents = deniedExists;
+ auditEvents.clear();
+ deniedExists = false;
}
- super.logAuthzAudit(auditEvent);
+ for (AuthzAuditEvent auditEvent : eventsToFlush) {
+ // If a deny exists, skip logging allowed results from the
same drained batch.
+ if (!deniedExistsForEvents || auditEvent.getAccessResult() ==
0) {
+ pendingAuditEvents.addLast(auditEvent);
+ }
+ }
+
+ while (!pendingAuditEvents.isEmpty()) {
+ AuthzAuditEvent auditEvent = pendingAuditEvents.peekFirst();
+ logAuditEvent(auditEvent);
+ // Remove only after the provider confirms delivery by
returning normally.
+ pendingAuditEvents.removeFirst();
+ }
}
}
- private void addAuthzAuditEvent(AuthzAuditEvent auditEvent) {
- if (auditEvent != null) {
+ protected void logAuditEvent(AuthzAuditEvent auditEvent) {
+ super.logAuthzAudit(auditEvent);
+ }
+
+ void addAuthzAuditEvent(AuthzAuditEvent auditEvent) {
+ if (auditEvent == null) {
+ return;
+ }
+ synchronized (auditBufferLock) {
auditEvents.add(auditEvent);
if (auditEvent.getAccessResult() == 0) {
@@ -259,6 +292,14 @@ public class RangerHiveAuditHandler extends
RangerDefaultAuditHandler {
}
}
+ int getPendingAuditEventCountForTest() {
+ synchronized (flushLock) {
+ synchronized (auditBufferLock) {
+ return pendingAuditEvents.size() + auditEvents.size();
+ }
+ }
+ }
+
private boolean skipFilterOperationAuditing(RangerAccessResult result) {
boolean ret = false;
RangerAccessRequest accessRequest = result.getAccessRequest();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditLogFlusher.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditLogFlusher.java
index e8afda11462..ffcd26d6667 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditLogFlusher.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditLogFlusher.java
@@ -17,12 +17,13 @@
package org.apache.doris.catalog.authorizer.ranger.hive;
-import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.TimerTask;
-@Slf4j
public class RangerHiveAuditLogFlusher extends TimerTask {
+ private static final Logger LOG =
LoggerFactory.getLogger(RangerHiveAuditLogFlusher.class);
private RangerHiveAuditHandler auditHandler;
public RangerHiveAuditLogFlusher(RangerHiveAuditHandler auditHandler) {
@@ -31,14 +32,12 @@ public class RangerHiveAuditLogFlusher extends TimerTask {
@Override
public void run() {
- while (true) {
+ try {
this.auditHandler.flushAudit();
-
- try {
- Thread.sleep(20000);
- } catch (InterruptedException e) {
- log.info("error ", e);
- }
+ } catch (Throwable t) {
+ // ScheduledThreadPoolExecutor suppresses every later fixed-rate
invocation when one run escapes with
+ // an exception. Keep the periodic flusher alive;
RangerHiveAuditHandler retains undelivered events.
+ LOG.warn("Failed to flush Ranger Hive audit events; will retry on
the next tick", t);
}
}
}
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 977840ef94f..a781fcc9eaa 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,11 +175,20 @@ public class CatalogFactory {
}
}
- // set some default properties if missing when creating catalog.
- // both replaying the creating logic will call this method.
- catalog.setDefaultPropsIfMissing(isReplay);
+ return finishCatalogCreation(catalog, isReplay);
+ }
+
+ static ExternalCatalog finishCatalogCreation(ExternalCatalog catalog,
boolean isReplay) throws DdlException {
+ // Set some default properties if missing when creating catalog.
+ // Both replaying the creating logic will call this method.
+ if (isReplay) {
+ catalog.setDefaultPropsIfMissing(true);
+ return catalog;
+ }
- if (!isReplay) {
+ boolean creationFinished = false;
+ try {
+ catalog.setDefaultPropsIfMissing(false);
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.
@@ -189,9 +198,12 @@ public class CatalogFactory {
LOG.warn("Failed to init access controller", e);
throw new DdlException("Failed to init access controller: " +
e.getMessage());
}
+ creationFinished = true;
+ return catalog;
+ } finally {
+ if (!creationFinished) {
+ catalog.onCreateFailure();
+ }
}
- return catalog;
}
}
-
-
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogIf.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogIf.java
index 1751ac107a6..e0ded1edc67 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogIf.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogIf.java
@@ -162,6 +162,11 @@ public interface CatalogIf<T extends DatabaseIf> {
// Called when catalog is dropped
void onClose();
+ // Called when catalog creation fails before the catalog is registered.
+ default void onCreateFailure() {
+ onClose();
+ }
+
String getComment();
default void setComment(String comment) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
index 0417e01f7f0..176d669f81d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
@@ -252,7 +252,7 @@ public class CatalogMgr implements Writable,
GsonPostProcessable {
try {
if (nameToCatalog.containsKey(catalog.getName())) {
// Close the already-constructed catalog to release connector
resources.
- catalog.onClose();
+ catalog.onCreateFailure();
if (ifNotExists) {
LOG.warn("Catalog {} is already exist.", catalogName);
return;
@@ -410,6 +410,7 @@ public class CatalogMgr implements Writable,
GsonPostProcessable {
* Modify the catalog property and write the meta log.
*/
public void alterCatalogProps(String catalogName, Map<String, String>
newProperties) throws UserException {
+ Runnable accessControllerCleanup = () -> { };
writeLock();
try {
CatalogIf catalog = nameToCatalog.get(catalogName);
@@ -424,10 +425,11 @@ public class CatalogMgr implements Writable,
GsonPostProcessable {
CatalogLog log = new CatalogLog();
log.setCatalogId(catalog.getId());
log.setNewProps(newProperties);
- replayAlterCatalogProps(log, oldProperties, false);
+ accessControllerCleanup = applyAlterCatalogProps(log,
oldProperties, false, true);
Env.getCurrentEnv().getEditLog().logCatalogLog(OperationType.OP_ALTER_CATALOG_PROPS,
log);
} finally {
writeUnlock();
+ accessControllerCleanup.run();
}
}
@@ -642,48 +644,60 @@ public class CatalogMgr implements Writable,
GsonPostProcessable {
*/
public void replayAlterCatalogProps(CatalogLog log, Map<String, String>
oldProperties, boolean isReplay)
throws DdlException {
+ Runnable accessControllerCleanup = () -> { };
writeLock();
try {
- CatalogIf catalog = idToCatalog.get(log.getCatalogId());
- if (catalog instanceof ExternalCatalog) {
- Map<String, String> newProps = log.getNewProps();
- if (!isReplay) {
- boolean tentativelyMutated = false;
- try {
- ExternalCatalog externalCatalog = (ExternalCatalog)
catalog;
- boolean validatedWithoutMutation =
externalCatalog.validatePropertiesBeforeUpdate(
- oldProperties, newProps);
- if (!validatedWithoutMutation) {
- externalCatalog.tryModifyCatalogProps(newProps);
- tentativelyMutated = true;
- externalCatalog.checkProperties();
- }
- } catch (Exception validationException) {
- // Only legacy validators publish a tentative
candidate. Detached validators
- // leave the live CatalogProperty untouched while
concurrent initialization runs.
- if (oldProperties != null && tentativelyMutated) {
- ((ExternalCatalog)
catalog).rollBackCatalogProps(oldProperties);
- }
- if (validationException instanceof DdlException) {
- throw (DdlException) validationException;
- }
- throw new DdlException("Invalid catalog properties: "
- + validationException.getMessage(),
validationException);
+ accessControllerCleanup = applyAlterCatalogProps(log,
oldProperties, isReplay, true);
+ } finally {
+ writeUnlock();
+ accessControllerCleanup.run();
+ }
+ }
+
+ private Runnable applyAlterCatalogProps(CatalogLog log, Map<String,
String> oldProperties,
+ boolean isReplay, boolean deferAccessControllerCleanup) throws
DdlException {
+ CatalogIf catalog = idToCatalog.get(log.getCatalogId());
+ if (catalog instanceof ExternalCatalog) {
+ Map<String, String> newProps = log.getNewProps();
+ if (!isReplay) {
+ boolean tentativelyMutated = false;
+ try {
+ ExternalCatalog externalCatalog = (ExternalCatalog)
catalog;
+ boolean validatedWithoutMutation =
externalCatalog.validatePropertiesBeforeUpdate(
+ oldProperties, newProps);
+ if (!validatedWithoutMutation) {
+ externalCatalog.tryModifyCatalogProps(newProps);
+ tentativelyMutated = true;
+ externalCatalog.checkProperties();
}
- } else {
- ((ExternalCatalog)
catalog).tryModifyCatalogProps(newProps);
- }
- if (newProps.containsKey(METADATA_REFRESH_INTERVAL_SEC)) {
- long catalogId = catalog.getId();
- Integer metadataRefreshIntervalSec =
Integer.valueOf(newProps.get(METADATA_REFRESH_INTERVAL_SEC));
- Integer[] sec = {metadataRefreshIntervalSec,
metadataRefreshIntervalSec};
-
Env.getCurrentEnv().getRefreshManager().addToRefreshMap(catalogId, sec);
+ } catch (Exception validationException) {
+ // Only legacy validators publish a tentative candidate.
Detached validators
+ // leave the live CatalogProperty untouched while
concurrent initialization runs.
+ if (oldProperties != null && tentativelyMutated) {
+ ((ExternalCatalog)
catalog).rollBackCatalogProps(oldProperties);
+ }
+ if (validationException instanceof DdlException) {
+ throw (DdlException) validationException;
+ }
+ throw new DdlException("Invalid catalog properties: "
+ + validationException.getMessage(),
validationException);
}
+ } else {
+ ((ExternalCatalog) catalog).tryModifyCatalogProps(newProps);
+ }
+ if (newProps.containsKey(METADATA_REFRESH_INTERVAL_SEC)) {
+ long catalogId = catalog.getId();
+ Integer metadataRefreshIntervalSec =
Integer.valueOf(newProps.get(METADATA_REFRESH_INTERVAL_SEC));
+ Integer[] sec = {metadataRefreshIntervalSec,
metadataRefreshIntervalSec};
+
Env.getCurrentEnv().getRefreshManager().addToRefreshMap(catalogId, sec);
+ }
+ if (deferAccessControllerCleanup) {
+ return ((ExternalCatalog) catalog)
+
.modifyCatalogPropsWithDeferredAccessControllerCleanup(log.getNewProps());
}
- catalog.modifyCatalogProps(log.getNewProps());
- } finally {
- writeUnlock();
}
+ catalog.modifyCatalogProps(log.getNewProps());
+ return () -> { };
}
public void unregisterExternalTable(String dbName, String tableName,
String catalogName, boolean ignoreIfExists)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
index 82048d0f0ed..86476b830da 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
@@ -496,7 +496,7 @@ public abstract class ExternalCatalog
}
// 3. create access controller
- Env.getCurrentEnv().getAccessManager().createAccessController(name,
className, acProperties, isDryRun);
+ Env.getCurrentEnv().getAccessManager().createAccessController(this,
className, acProperties, isDryRun);
}
/**
@@ -599,12 +599,28 @@ public abstract class ExternalCatalog
* and reloaded during the refresh process.
*/
public void resetToUninitialized(boolean invalidCache) {
+ resetToUninitialized(invalidCache, false);
+ }
+
+ private Runnable resetToUninitialized(boolean invalidCache, boolean
deferAccessControllerCleanup) {
+ Runnable accessControllerCleanup;
synchronized (this) {
this.objectCreated = false;
this.initialized = false;
- onClose();
+ accessControllerCleanup = detachAccessController();
+ closeResourcesQuietly("resetting catalog");
+ }
+ try {
+ onRefreshCache(invalidCache);
+ } catch (RuntimeException | Error e) {
+ accessControllerCleanup.run();
+ throw e;
}
- onRefreshCache(invalidCache);
+ if (!deferAccessControllerCleanup) {
+ accessControllerCleanup.run();
+ return () -> { };
+ }
+ return accessControllerCleanup;
}
/**
@@ -832,6 +848,25 @@ public abstract class ExternalCatalog
notifyPropertiesUpdated(props);
}
+ /**
+ * Apply property changes while atomically detaching the old access
controller, but return its potentially
+ * blocking close operation to CatalogMgr so it can run after releasing
the global catalog write lock.
+ */
+ public Runnable
modifyCatalogPropsWithDeferredAccessControllerCleanup(Map<String, String>
props) {
+ catalogProperty.modifyCatalogProps(props);
+ Runnable accessControllerCleanup = resetToUninitialized(false, true);
+ try {
+ String schemaCacheTtl =
props.getOrDefault(SCHEMA_CACHE_TTL_SECOND, null);
+ if (java.util.Objects.nonNull(schemaCacheTtl)) {
+ Env.getCurrentEnv().getExtMetaCacheMgr().removeCatalog(id);
+ }
+ return accessControllerCleanup;
+ } catch (RuntimeException | Error e) {
+ accessControllerCleanup.run();
+ throw e;
+ }
+ }
+
public void tryModifyCatalogProps(Map<String, String> props) {
catalogProperty.modifyCatalogProps(props);
}
@@ -849,8 +884,29 @@ public abstract class ExternalCatalog
}
@Override
- public void onClose() {
- removeAccessController();
+ public final void onClose() {
+ Runnable accessControllerCleanup = detachAccessController();
+ try {
+ closeResourcesQuietly("registered catalog");
+ } finally {
+ accessControllerCleanup.run();
+ }
+ }
+
+ @Override
+ public final void onCreateFailure() {
+ closeResourcesQuietly("unregistered catalog");
+ }
+
+ private void closeResourcesQuietly(String lifecycleStage) {
+ try {
+ closeResources();
+ } catch (Throwable t) {
+ LOG.warn("Failed to close resources for {} {}", lifecycleStage,
name, t);
+ }
+ }
+
+ protected void closeResources() {
if (threadPoolWithPreAuth != null) {
ThreadPoolManager.shutdownExecutorService(threadPoolWithPreAuth);
}
@@ -862,8 +918,8 @@ public abstract class ExternalCatalog
}
}
- private void removeAccessController() {
- Env.getCurrentEnv().getAccessManager().removeAccessController(name);
+ private Runnable detachAccessController() {
+ return
Env.getCurrentEnv().getAccessManager().detachAccessController(name, id);
}
/**
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcOceanBaseClient.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcOceanBaseClient.java
index 004e7c80f3d..8dda7721335 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcOceanBaseClient.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcOceanBaseClient.java
@@ -57,6 +57,7 @@ public class JdbcOceanBaseClient extends JdbcClient {
throw new JdbcClientException("Failed to initialize
JdbcOceanBaseClient: %s", e.getMessage());
} finally {
close(rs, stmt, conn);
+ closeClient();
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalCatalog.java
index e228fca35a7..657d8dfbf8a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalCatalog.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalCatalog.java
@@ -1513,21 +1513,29 @@ public class PluginDrivenExternalCatalog extends
ExternalCatalog {
}
try {
context.close();
- } catch (IOException e) {
+ } catch (Throwable e) {
LOG.warn("Failed to close connector context filesystem for catalog
{}", name, e);
}
}
@Override
- public void onClose() {
- super.onClose();
- if (connector != null) {
+ protected void closeResources() {
+ try {
+ super.closeResources();
+ } catch (Throwable e) {
+ LOG.warn("Failed to close common resources for plugin-driven
catalog {}", name, e);
+ }
+
+ // Detach every stage before invoking external code. A throwing
connector must not remain reachable
+ // for another close attempt or prevent the connector-context stage
from running.
+ Connector connectorToClose = connector;
+ connector = null;
+ if (connectorToClose != null) {
try {
- connector.close();
- } catch (IOException e) {
+ connectorToClose.close();
+ } catch (Throwable e) {
LOG.warn("Failed to close connector for catalog {}", name, e);
}
- connector = null;
}
// Close the shared context's cached engine FileSystem AFTER the
connector(s) release their borrowed
// reference to it. No-op when no FS was ever built (e.g. non-hive
plugin catalogs never call
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/main/java/org/apache/doris/mysql/privilege/AccessControllerManager.java
b/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/AccessControllerManager.java
index 3cfa387efa7..1268341600d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/AccessControllerManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/AccessControllerManager.java
@@ -61,8 +61,9 @@ public class AccessControllerManager {
private Auth auth;
// Default access controller instance used for handling cases where no
specific controller is specified
private CatalogAccessController defaultAccessController;
- // Map that stores the mapping between catalogs and their corresponding
access controllers
- private Map<String, CatalogAccessController> ctlToCtlAccessController =
Maps.newConcurrentMap();
+ // A catalog name can be reused after DROP. Keep the catalog id next to
the controller so cleanup from
+ // an old catalog generation can never remove or close the replacement
generation's controller.
+ private Map<String, CatalogAccessControllerEntry> ctlToCtlAccessController
= Maps.newConcurrentMap();
// Cache of loaded access controller factories for quick creation of new
access controllers
private ConcurrentHashMap<String, AccessControllerFactory>
accessControllerFactoriesCache
= new ConcurrentHashMap<>();
@@ -74,7 +75,24 @@ public class AccessControllerManager {
loadAccessControllerPlugins();
String accessControllerName = Config.access_controller_type;
this.defaultAccessController =
loadAccessControllerOrThrow(accessControllerName);
- ctlToCtlAccessController.put(InternalCatalog.INTERNAL_CATALOG_NAME,
defaultAccessController);
+ ctlToCtlAccessController.put(InternalCatalog.INTERNAL_CATALOG_NAME,
+ new CatalogAccessControllerEntry(
+ InternalCatalog.INTERNAL_CATALOG_ID,
defaultAccessController, false));
+ }
+
+ private static final class CatalogAccessControllerEntry {
+ private final long catalogId;
+ private final CatalogAccessController accessController;
+ // The default controller is shared with the internal catalog. Catalog
aliases must detach it but never
+ // close it when an external catalog is reset or dropped.
+ private final boolean owned;
+
+ private CatalogAccessControllerEntry(
+ long catalogId, CatalogAccessController accessController,
boolean owned) {
+ this.catalogId = catalogId;
+ this.accessController = accessController;
+ this.owned = owned;
+ }
}
private CatalogAccessController loadAccessControllerOrThrow(String
accessControllerName) {
@@ -116,26 +134,66 @@ public class AccessControllerManager {
}
public CatalogAccessController getAccessControllerOrDefault(String ctl) {
- CatalogAccessController catalogAccessController =
ctlToCtlAccessController.get(ctl);
- if (catalogAccessController != null) {
- return catalogAccessController;
+ if (InternalCatalog.INTERNAL_CATALOG_NAME.equals(ctl)) {
+ return defaultAccessController;
}
CatalogIf catalog =
Env.getCurrentEnv().getCatalogMgr().getCatalog(ctl);
if (catalog != null && catalog instanceof ExternalCatalog) {
+ CatalogAccessControllerEntry entry =
ctlToCtlAccessController.get(ctl);
+ if (entry != null && entry.catalogId == catalog.getId()) {
+ return entry.accessController;
+ }
lazyLoadCtlAccessController((ExternalCatalog) catalog);
- return ctlToCtlAccessController.get(ctl);
+ entry = ctlToCtlAccessController.get(ctl);
+ if (entry != null && entry.catalogId == catalog.getId()) {
+ return entry.accessController;
+ }
}
return defaultAccessController;
}
- private synchronized void lazyLoadCtlAccessController(ExternalCatalog
catalog) {
- if (ctlToCtlAccessController.containsKey(catalog.getName())) {
- return;
+ private void lazyLoadCtlAccessController(ExternalCatalog catalog) {
+ CatalogAccessControllerEntry staleEntry = null;
+ synchronized (this) {
+ if (!isCurrentCatalog(catalog)) {
+ return;
+ }
+ CatalogAccessControllerEntry entry =
ctlToCtlAccessController.get(catalog.getName());
+ if (entry != null && entry.catalogId == catalog.getId()) {
+ return;
+ }
+ if (entry != null &&
ctlToCtlAccessController.remove(catalog.getName(), entry)) {
+ staleEntry = entry;
+ }
}
+ closeEntry(catalog.getName(), staleEntry);
+
catalog.initAccessController(false);
- if (!ctlToCtlAccessController.containsKey(catalog.getName())) {
- ctlToCtlAccessController.put(catalog.getName(),
defaultAccessController);
+
+ CatalogAccessControllerEntry displaced = null;
+ boolean stillCurrent;
+ synchronized (this) {
+ stillCurrent = isCurrentCatalog(catalog);
+ if (stillCurrent) {
+ CatalogAccessControllerEntry entry =
ctlToCtlAccessController.get(catalog.getName());
+ if (entry == null || entry.catalogId != catalog.getId()) {
+ displaced = ctlToCtlAccessController.put(catalog.getName(),
+ new CatalogAccessControllerEntry(catalog.getId(),
defaultAccessController, false));
+ }
+ }
+ }
+ closeEntry(catalog.getName(), displaced);
+ if (!stillCurrent) {
+ // A DROP can complete while initAccessController() is
constructing the plugin. The custom
+ // publication path performs the same post-publication check; this
also covers the fallback path.
+ removeAccessController(catalog.getName(), catalog.getId());
+ return;
+ }
+ // If DROP won immediately after the synchronized publication, its
onClose() removes this id. If DROP
+ // already completed before publication, this final identity check
removes the orphan ourselves.
+ if (!isCurrentCatalog(catalog)) {
+ removeAccessController(catalog.getName(), catalog.getId());
}
}
@@ -143,17 +201,45 @@ public class AccessControllerManager {
return ctlToCtlAccessController.containsKey(ctl);
}
- public void createAccessController(String ctl, String acFactoryClassName,
Map<String, String> prop,
+ public void createAccessController(ExternalCatalog catalog, String
acFactoryClassName, Map<String, String> prop,
boolean isDryRun) {
String pluginIdentifier =
getPluginIdentifierForAccessController(acFactoryClassName);
CatalogAccessController accessController =
accessControllerFactoriesCache.get(pluginIdentifier)
.createAccessController(prop);
- if (!isDryRun) {
- ctlToCtlAccessController.put(ctl, accessController);
- LOG.info("create access controller {} for catalog {}",
acFactoryClassName, ctl);
+ if (isDryRun) {
+ closeAccessController(catalog.getName(), accessController);
+ return;
+ }
+
+ CatalogAccessControllerEntry displaced = null;
+ boolean installed = false;
+ synchronized (this) {
+ if (isCurrentCatalog(catalog)) {
+ CatalogAccessControllerEntry current =
ctlToCtlAccessController.get(catalog.getName());
+ if (current == null || current.catalogId != catalog.getId()) {
+ displaced = ctlToCtlAccessController.put(catalog.getName(),
+ new CatalogAccessControllerEntry(catalog.getId(),
accessController, true));
+ installed = true;
+ }
+ }
+ }
+ closeEntry(catalog.getName(), displaced);
+ if (!installed) {
+ closeAccessController(catalog.getName(), accessController);
+ return;
+ }
+ LOG.info("create access controller {} for catalog {}:{}",
+ acFactoryClassName, catalog.getName(), catalog.getId());
+ if (!isCurrentCatalog(catalog)) {
+ removeAccessController(catalog.getName(), catalog.getId());
}
}
+ private boolean isCurrentCatalog(ExternalCatalog catalog) {
+ CatalogIf currentCatalog =
Env.getCurrentEnv().getCatalogMgr().getCatalog(catalog.getName());
+ return currentCatalog == catalog && currentCatalog.getId() ==
catalog.getId();
+ }
+
private String getPluginIdentifierForAccessController(String acClassName) {
String pluginIdentifier = null;
if (accessControllerClassNameMapping.containsKey(acClassName)) {
@@ -168,14 +254,41 @@ public class AccessControllerManager {
return pluginIdentifier;
}
- public void removeAccessController(String ctl) {
+ public void removeAccessController(String ctl, long catalogId) {
+ detachAccessController(ctl, catalogId).run();
+ }
+
+ /**
+ * Atomically detach the controller owned by one catalog generation. The
returned cleanup can be executed
+ * after the caller releases CatalogMgr's global lock, so a slow plugin
close never blocks unrelated DDL.
+ */
+ public Runnable detachAccessController(String ctl, long catalogId) {
if (StringUtils.isBlank(ctl)) {
+ return () -> { };
+ }
+ CatalogAccessControllerEntry entry = ctlToCtlAccessController.get(ctl);
+ if (entry == null || entry.catalogId != catalogId ||
!ctlToCtlAccessController.remove(ctl, entry)) {
+ return () -> { };
+ }
+ LOG.info("detach access controller for catalog {}:{}", ctl, catalogId);
+ return () -> closeEntry(ctl, entry);
+ }
+
+ private void closeEntry(String ctl, CatalogAccessControllerEntry entry) {
+ if (entry == null || !entry.owned || entry.accessController ==
defaultAccessController) {
return;
}
- if (ctlToCtlAccessController.containsKey(ctl)) {
- ctlToCtlAccessController.remove(ctl);
+ closeAccessController(ctl, entry.accessController);
+ }
+
+ private void closeAccessController(String ctl, CatalogAccessController
accessController) {
+ try {
+ accessController.close();
+ } catch (Throwable e) {
+ // Access-controller plugins are external code. A faulty cleanup
must not prevent the catalog
+ // lifecycle from releasing its own resources.
+ LOG.warn("Failed to close access controller for catalog {}", ctl,
e);
}
- LOG.info("remove access controller for catalog {}", ctl);
}
public Auth getAuth() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/CatalogAccessController.java
b/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/CatalogAccessController.java
index df7605256dd..8e548b7463f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/CatalogAccessController.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/CatalogAccessController.java
@@ -26,6 +26,9 @@ import java.util.Optional;
import java.util.Set;
public interface CatalogAccessController {
+ default void close() {
+ }
+
// ==== Catalog ====
default boolean checkCtlPriv(boolean hasGlobal, UserIdentity currentUser,
String ctl, PrivPredicate wanted) {
if (hasGlobal) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditLogFlusherTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditLogFlusherTest.java
new file mode 100644
index 00000000000..92e9440aaf3
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/authorizer/ranger/hive/RangerHiveAuditLogFlusherTest.java
@@ -0,0 +1,139 @@
+// 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.catalog.authorizer.ranger.hive;
+
+import org.apache.ranger.audit.model.AuthzAuditEvent;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class RangerHiveAuditLogFlusherTest {
+
+ @Test
+ public void testRunFlushesOnceAndReturns() {
+ RangerHiveAuditHandler auditHandler =
Mockito.mock(RangerHiveAuditHandler.class);
+
+ new RangerHiveAuditLogFlusher(auditHandler).run();
+
+ Mockito.verify(auditHandler).flushAudit();
+ }
+
+ @Test
+ public void testRunContainsProviderFailureSoLaterTicksContinue() {
+ RangerHiveAuditHandler auditHandler =
Mockito.mock(RangerHiveAuditHandler.class);
+ Mockito.doThrow(new RuntimeException("provider unavailable"))
+ .doNothing().when(auditHandler).flushAudit();
+ RangerHiveAuditLogFlusher flusher = new
RangerHiveAuditLogFlusher(auditHandler);
+
+ flusher.run();
+ flusher.run();
+
+ Mockito.verify(auditHandler, Mockito.times(2)).flushAudit();
+ }
+
+ @Test
+ public void testProducerCanEnqueueWhileProviderDeliveryIsBlocked() throws
Exception {
+ RecordingAuditHandler auditHandler = new RecordingAuditHandler(false,
true);
+ AuthzAuditEvent first = allowedEvent();
+ AuthzAuditEvent second = allowedEvent();
+ auditHandler.addAuthzAuditEvent(first);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+ try {
+ Future<?> flush = executor.submit(auditHandler::flushAudit);
+ Assert.assertTrue(auditHandler.deliveryStarted.await(10,
TimeUnit.SECONDS));
+
+ Future<?> producer = executor.submit(() ->
auditHandler.addAuthzAuditEvent(second));
+ producer.get(2, TimeUnit.SECONDS);
+
+ auditHandler.allowDelivery.countDown();
+ flush.get(10, TimeUnit.SECONDS);
+ auditHandler.flushAudit();
+
+ Assert.assertEquals(List.of(first, second),
auditHandler.delivered);
+ } finally {
+ auditHandler.allowDelivery.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testFailedDeliveryIsRetriedBeforeLaterEvents() {
+ RecordingAuditHandler auditHandler = new RecordingAuditHandler(true,
false);
+ AuthzAuditEvent first = allowedEvent();
+ AuthzAuditEvent second = allowedEvent();
+ auditHandler.addAuthzAuditEvent(first);
+
+ Assert.assertThrows(RuntimeException.class, auditHandler::flushAudit);
+ Assert.assertEquals(1,
auditHandler.getPendingAuditEventCountForTest());
+
+ auditHandler.addAuthzAuditEvent(second);
+ auditHandler.flushAudit();
+
+ Assert.assertEquals(List.of(first, second), auditHandler.delivered);
+ Assert.assertEquals(0,
auditHandler.getPendingAuditEventCountForTest());
+ }
+
+ private static AuthzAuditEvent allowedEvent() {
+ AuthzAuditEvent event = new AuthzAuditEvent();
+ event.setAccessResult((short) 1);
+ return event;
+ }
+
+ private static class RecordingAuditHandler extends RangerHiveAuditHandler {
+ private final AtomicBoolean failNext;
+ private final boolean blockFirstDelivery;
+ private final AtomicBoolean firstDelivery = new AtomicBoolean(true);
+ private final CountDownLatch deliveryStarted = new CountDownLatch(1);
+ private final CountDownLatch allowDelivery = new CountDownLatch(1);
+ private final List<AuthzAuditEvent> delivered = new
CopyOnWriteArrayList<>();
+
+ RecordingAuditHandler(boolean failFirstDelivery, boolean
blockFirstDelivery) {
+ this.failNext = new AtomicBoolean(failFirstDelivery);
+ this.blockFirstDelivery = blockFirstDelivery;
+ }
+
+ @Override
+ protected void logAuditEvent(AuthzAuditEvent auditEvent) {
+ if (blockFirstDelivery && firstDelivery.compareAndSet(true,
false)) {
+ deliveryStarted.countDown();
+ try {
+ if (!allowDelivery.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("timed out waiting to
release audit provider");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+ if (failNext.compareAndSet(true, false)) {
+ throw new RuntimeException("provider unavailable");
+ }
+ delivered.add(auditEvent);
+ }
+ }
+}
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..c03f9c4ade7
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogFactoryTest.java
@@ -0,0 +1,90 @@
+// 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.common.jmockit.Deencapsulation;
+import org.apache.doris.connector.DefaultConnectorContext;
+import org.apache.doris.connector.spi.Connector;
+import org.apache.doris.datasource.plugin.PluginDrivenExternalCatalog;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+
+public class CatalogFactoryTest {
+
+ @Test
+ public void testCloseCatalogWhenCreateValidationFails() throws Exception {
+ Connector connector = Mockito.mock(Connector.class);
+ PluginDrivenExternalCatalog catalog = Mockito.spy(new
PluginDrivenExternalCatalog(
+ 1L, "failed_catalog", null, new HashMap<>(), "", connector));
+ Mockito.doThrow(new DdlException("validation
failed")).when(catalog).checkWhenCreating();
+
+ DdlException exception = Assert.assertThrows(
+ DdlException.class, () ->
CatalogFactory.finishCatalogCreation(catalog, false));
+
+ Assert.assertTrue(exception.getMessage().endsWith("validation
failed"));
+ Mockito.verify(connector).close();
+ }
+
+ @Test
+ public void testPreserveValidationFailureWhenCleanupFails() throws
Exception {
+ Connector connector = Mockito.mock(Connector.class);
+ Mockito.doThrow(new RuntimeException("cleanup
failed")).when(connector).close();
+ PluginDrivenExternalCatalog catalog = new PluginDrivenExternalCatalog(
+ 2L, "failed_catalog", null, new HashMap<>(), "", connector) {
+ @Override
+ public void checkWhenCreating() throws DdlException {
+ throw new DdlException("primary validation failure");
+ }
+ };
+
+ DdlException exception = Assert.assertThrows(
+ DdlException.class, () ->
CatalogFactory.finishCatalogCreation(catalog, false));
+
+ Assert.assertTrue(exception.getMessage().endsWith("primary validation
failure"));
+ }
+
+ @Test
+ public void
testRuntimeConnectorFailureDoesNotSkipContextCleanupOrRetryConnector() throws
Exception {
+ Connector connector = Mockito.mock(Connector.class);
+ DefaultConnectorContext connectorContext =
Mockito.mock(DefaultConnectorContext.class);
+ Mockito.doThrow(new RuntimeException("connector cleanup
failed")).when(connector).close();
+ TestablePluginCatalog catalog = new TestablePluginCatalog(connector);
+ Deencapsulation.setField(catalog, "connectorContext",
connectorContext);
+
+ catalog.closeResourcesForTest();
+ catalog.closeResourcesForTest();
+
+ Mockito.verify(connector).close();
+ Mockito.verify(connectorContext).close();
+ }
+
+ private static class TestablePluginCatalog extends
PluginDrivenExternalCatalog {
+ TestablePluginCatalog(Connector connector) {
+ super(3L, "registered_catalog", null, new HashMap<>(), "",
connector);
+ }
+
+ void closeResourcesForTest() {
+ closeResources();
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java
index fb43b3d99c8..e16cab34a80 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogMgrTest.java
@@ -113,6 +113,66 @@ public class CatalogMgrTest {
}
}
+ @Test
+ void testAlterControllerCleanupRunsOutsideCatalogWriteLock() throws
Exception {
+ CatalogMgr catalogMgr = new CatalogMgr();
+ ExternalCatalog blockingCatalog = Mockito.mock(ExternalCatalog.class);
+ ExternalCatalog otherCatalog = Mockito.mock(ExternalCatalog.class);
+ Mockito.when(blockingCatalog.getId()).thenReturn(44L);
+ Mockito.when(otherCatalog.getId()).thenReturn(45L);
+
Mockito.when(blockingCatalog.validatePropertiesBeforeUpdate(Mockito.anyMap(),
Mockito.anyMap()))
+ .thenReturn(true);
+
Mockito.when(otherCatalog.validatePropertiesBeforeUpdate(Mockito.anyMap(),
Mockito.anyMap()))
+ .thenReturn(true);
+ addCatalog(catalogMgr, blockingCatalog);
+ addCatalog(catalogMgr, otherCatalog);
+
+ CountDownLatch cleanupStarted = new CountDownLatch(1);
+ CountDownLatch allowCleanup = new CountDownLatch(1);
+
Mockito.when(blockingCatalog.modifyCatalogPropsWithDeferredAccessControllerCleanup(Mockito.anyMap()))
+ .thenReturn(() -> {
+ cleanupStarted.countDown();
+ try {
+ if (!allowCleanup.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("timed out waiting
to release controller cleanup");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ });
+
Mockito.when(otherCatalog.modifyCatalogPropsWithDeferredAccessControllerCleanup(Mockito.anyMap()))
+ .thenReturn(() -> { });
+
+ CatalogLog blockingLog = new CatalogLog();
+ blockingLog.setCatalogId(44L);
+ blockingLog.setNewProps(ImmutableMap.of("k", "v1"));
+ CatalogLog otherLog = new CatalogLog();
+ otherLog.setCatalogId(45L);
+ otherLog.setNewProps(ImmutableMap.of("k", "v2"));
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+ try {
+ Future<?> blockingAlter = executor.submit(() -> {
+ catalogMgr.replayAlterCatalogProps(blockingLog,
Collections.emptyMap(), false);
+ return null;
+ });
+ Assertions.assertTrue(cleanupStarted.await(10, TimeUnit.SECONDS));
+
+ Future<?> unrelatedAlter = executor.submit(() -> {
+ catalogMgr.replayAlterCatalogProps(otherLog,
Collections.emptyMap(), false);
+ return null;
+ });
+ unrelatedAlter.get(2, TimeUnit.SECONDS);
+
+ allowCleanup.countDown();
+ blockingAlter.get(10, TimeUnit.SECONDS);
+ } finally {
+ allowCleanup.countDown();
+ executor.shutdownNow();
+ }
+ }
+
private static class LatchingValidationCatalog extends ExternalCatalog {
private final CountDownLatch validationStarted = new CountDownLatch(1);
private final CountDownLatch initializationReadProperties = new
CountDownLatch(1);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogDeadlockTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogDeadlockTest.java
index c6a9ce8841a..c9a46b4c1dd 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogDeadlockTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogDeadlockTest.java
@@ -32,12 +32,15 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
public class ExternalCatalogDeadlockTest {
+ private static final long DEADLOCK_DETECTION_TIMEOUT_SECONDS = 5;
+ private static final long LOADER_PERMISSION_TIMEOUT_SECONDS = 30;
@Test
public void testResetToUninitializedShouldNotDeadlockWithCacheLoader()
throws Exception {
DeadlockCatalog catalog = new DeadlockCatalog();
CountDownLatch loaderEntered = new CountDownLatch(1);
CountDownLatch allowLoaderToTouchCatalog = new CountDownLatch(1);
+ CountDownLatch refreshReady = new CountDownLatch(1);
AtomicReference<Throwable> backgroundFailure = new AtomicReference<>();
// The loader holds Caffeine's per-key lock before it calls back into
the catalog.
@@ -53,7 +56,7 @@ public class ExternalCatalogDeadlockTest {
"deadlock-cache-loader");
queryThread.setDaemon(true);
queryThread.start();
- Assertions.assertTrue(loaderEntered.await(5, TimeUnit.SECONDS));
+
Assertions.assertTrue(loaderEntered.await(DEADLOCK_DETECTION_TIMEOUT_SECONDS,
TimeUnit.SECONDS));
Thread refreshThread = new Thread(
() -> runQuietly(backgroundFailure, () -> {
@@ -62,11 +65,13 @@ public class ExternalCatalogDeadlockTest {
allowLoaderToTouchCatalog.countDown();
cache.invalidate("deadlock-key");
});
+ refreshReady.countDown();
catalog.resetToUninitialized(true);
}),
"deadlock-catalog-refresh");
refreshThread.setDaemon(true);
refreshThread.start();
+
Assertions.assertTrue(refreshReady.await(DEADLOCK_DETECTION_TIMEOUT_SECONDS,
TimeUnit.SECONDS));
assertNoDeadlock(queryThread, refreshThread, backgroundFailure);
}
@@ -74,8 +79,8 @@ public class ExternalCatalogDeadlockTest {
private static void assertNoDeadlock(Thread queryThread, Thread
refreshThread,
AtomicReference<Throwable> backgroundFailure) throws Exception {
long[] deadlockedThreads = waitForDeadlock(queryThread, refreshThread);
- queryThread.join(TimeUnit.SECONDS.toMillis(5));
- refreshThread.join(TimeUnit.SECONDS.toMillis(5));
+
queryThread.join(TimeUnit.SECONDS.toMillis(DEADLOCK_DETECTION_TIMEOUT_SECONDS));
+
refreshThread.join(TimeUnit.SECONDS.toMillis(DEADLOCK_DETECTION_TIMEOUT_SECONDS));
Assertions.assertNull(backgroundFailure.get(), "unexpected background
failure: " + backgroundFailure.get());
Assertions.assertNull(deadlockedThreads,
String.format("detected deadlock between threads %s and %s",
@@ -85,7 +90,7 @@ public class ExternalCatalogDeadlockTest {
}
private static void awaitLatch(CountDownLatch latch) throws
InterruptedException {
- Assertions.assertTrue(latch.await(5, TimeUnit.SECONDS));
+ Assertions.assertTrue(latch.await(LOADER_PERMISSION_TIMEOUT_SECONDS,
TimeUnit.SECONDS));
}
private static void runQuietly(AtomicReference<Throwable> failure,
ThrowingRunnable task) {
@@ -98,7 +103,7 @@ public class ExternalCatalogDeadlockTest {
private static long[] waitForDeadlock(Thread queryThread, Thread
refreshThread) throws InterruptedException {
ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean();
- for (int i = 0; i < 100; i++) {
+ for (int i = 0; i < DEADLOCK_DETECTION_TIMEOUT_SECONDS * 20; i++) {
long[] deadlockedThreads = threadMxBean.findDeadlockedThreads();
if (deadlockedThreads != null
&& contains(deadlockedThreads, queryThread.getId())
@@ -132,7 +137,7 @@ public class ExternalCatalogDeadlockTest {
}
@Override
- public void onClose() {
+ protected void closeResources() {
}
@Override
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/jdbc/client/JdbcOceanBaseClientTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/jdbc/client/JdbcOceanBaseClientTest.java
new file mode 100644
index 00000000000..b2632b8fc10
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/jdbc/client/JdbcOceanBaseClientTest.java
@@ -0,0 +1,105 @@
+// 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.jdbc.client;
+
+import com.zaxxer.hikari.HikariDataSource;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.MockedConstruction;
+import org.mockito.Mockito;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+public class JdbcOceanBaseClientTest {
+ private Connection connection;
+ private Statement statement;
+ private ResultSet resultSet;
+
+ @Before
+ public void setUp() throws Exception {
+ connection = Mockito.mock(Connection.class);
+ statement = Mockito.mock(Statement.class);
+ resultSet = Mockito.mock(ResultSet.class);
+ Mockito.when(connection.createStatement()).thenReturn(statement);
+ Mockito.when(statement.executeQuery("SHOW VARIABLES LIKE
'ob_compatibility_mode'")).thenReturn(resultSet);
+ }
+
+ @Test
+ public void testCloseTemporaryDataSourceAfterCreatingClient() throws
Exception {
+ Mockito.when(resultSet.next()).thenReturn(true);
+ Mockito.when(resultSet.getString(2)).thenReturn("MYSQL");
+
+ try (MockedConstruction<HikariDataSource> mockedDataSources =
mockDataSources()) {
+ JdbcOceanBaseClient oceanBaseClient = new
JdbcOceanBaseClient(createConfig());
+ JdbcClient client = oceanBaseClient.createClient(createConfig());
+
+ Assert.assertTrue(client instanceof JdbcMySQLClient);
+ Assert.assertEquals(2, mockedDataSources.constructed().size());
+ HikariDataSource temporaryDataSource =
mockedDataSources.constructed().get(0);
+ HikariDataSource clientDataSource =
mockedDataSources.constructed().get(1);
+ assertTemporaryResourcesClosed(temporaryDataSource);
+ Mockito.verify(clientDataSource, Mockito.never()).close();
+
+ client.closeClient();
+ Mockito.verify(clientDataSource).close();
+ }
+ }
+
+ @Test
+ public void testCloseTemporaryDataSourceWhenCompatibilityModeIsMissing()
throws Exception {
+ Mockito.when(resultSet.next()).thenReturn(false);
+
+ try (MockedConstruction<HikariDataSource> mockedDataSources =
mockDataSources()) {
+ JdbcOceanBaseClient oceanBaseClient = new
JdbcOceanBaseClient(createConfig());
+
+ JdbcClientException exception = Assert.assertThrows(
+ JdbcClientException.class, () ->
oceanBaseClient.createClient(createConfig()));
+
+ Assert.assertEquals("Failed to determine OceanBase compatibility
mode", exception.getMessage());
+ Assert.assertEquals(1, mockedDataSources.constructed().size());
+
assertTemporaryResourcesClosed(mockedDataSources.constructed().get(0));
+ }
+ }
+
+ private MockedConstruction<HikariDataSource> mockDataSources() {
+ return Mockito.mockConstruction(HikariDataSource.class, (mock,
context) ->
+ Mockito.when(mock.getConnection()).thenReturn(connection));
+ }
+
+ private JdbcClientConfig createConfig() {
+ return new JdbcClientConfig()
+ .setCatalog("oceanbase_catalog")
+ .setUser("user")
+ .setPassword("password")
+ .setJdbcUrl("jdbc:oceanbase://localhost:2881/test")
+ .setDriverUrl("file:///tmp/oceanbase-jdbc.jar")
+ .setDriverClass("com.oceanbase.jdbc.Driver");
+ }
+
+ private void assertTemporaryResourcesClosed(HikariDataSource
temporaryDataSource) throws Exception {
+ InOrder inOrder = Mockito.inOrder(resultSet, statement, connection,
temporaryDataSource);
+ inOrder.verify(resultSet).close();
+ inOrder.verify(statement).close();
+ inOrder.verify(connection).close();
+ inOrder.verify(temporaryDataSource).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();
+ }
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mysql/privilege/AccessControllerManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mysql/privilege/AccessControllerManagerTest.java
index bebe5e98ef3..a6b2514b43a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mysql/privilege/AccessControllerManagerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mysql/privilege/AccessControllerManagerTest.java
@@ -23,6 +23,7 @@ import org.apache.doris.common.Config;
import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.ExternalCatalog;
import com.google.common.collect.ImmutableMap;
import org.junit.After;
@@ -32,6 +33,9 @@ import org.junit.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+
public class AccessControllerManagerTest {
private boolean originalSkipCatalogPrivCheck;
@@ -193,6 +197,164 @@ public class AccessControllerManagerTest {
}
}
+ @Test
+ public void testDryRunClosesTemporaryAccessController() {
+ CatalogAccessController defaultAccessController =
Mockito.mock(CatalogAccessController.class);
+ CatalogAccessController temporaryAccessController =
Mockito.mock(CatalogAccessController.class);
+ AccessControllerFactory factory =
Mockito.mock(AccessControllerFactory.class);
+ AccessControllerManager accessControllerManager =
createAccessControllerManager(defaultAccessController);
+ ConcurrentHashMap<String, AccessControllerFactory> factories =
+ Deencapsulation.getField(accessControllerManager,
"accessControllerFactoriesCache");
+ factories.put("test-controller", factory);
+
Mockito.when(factory.createAccessController(ImmutableMap.of())).thenReturn(temporaryAccessController);
+ ExternalCatalog catalog = mockCatalog("test_catalog", 1L);
+
+ accessControllerManager.createAccessController(
+ catalog, "test-controller", ImmutableMap.of(), true);
+
+ Mockito.verify(temporaryAccessController).close();
+
Assert.assertFalse(accessControllerManager.checkIfAccessControllerExist("test_catalog"));
+ }
+
+ @Test
+ public void testRemoveClosesRegisteredAccessController() {
+ CatalogAccessController defaultAccessController =
Mockito.mock(CatalogAccessController.class);
+ CatalogAccessController registeredAccessController =
Mockito.mock(CatalogAccessController.class);
+ AccessControllerFactory factory =
Mockito.mock(AccessControllerFactory.class);
+ AccessControllerManager accessControllerManager =
createAccessControllerManager(defaultAccessController);
+ ConcurrentHashMap<String, AccessControllerFactory> factories =
+ Deencapsulation.getField(accessControllerManager,
"accessControllerFactoriesCache");
+ factories.put("test-controller", factory);
+
Mockito.when(factory.createAccessController(ImmutableMap.of())).thenReturn(registeredAccessController);
+ ExternalCatalog catalog = mockCatalog("test_catalog", 1L);
+
+ withCurrentCatalog(catalog, () ->
accessControllerManager.createAccessController(
+ catalog, "test-controller", ImmutableMap.of(), false));
+ accessControllerManager.removeAccessController("test_catalog",
catalog.getId());
+
+ Mockito.verify(registeredAccessController).close();
+
Assert.assertFalse(accessControllerManager.checkIfAccessControllerExist("test_catalog"));
+ }
+
+ @Test
+ public void testRemoveContinuesWhenAccessControllerCloseFails() {
+ CatalogAccessController defaultAccessController =
Mockito.mock(CatalogAccessController.class);
+ CatalogAccessController registeredAccessController =
Mockito.mock(CatalogAccessController.class);
+ AccessControllerFactory factory =
Mockito.mock(AccessControllerFactory.class);
+ AccessControllerManager accessControllerManager =
createAccessControllerManager(defaultAccessController);
+ ConcurrentHashMap<String, AccessControllerFactory> factories =
+ Deencapsulation.getField(accessControllerManager,
"accessControllerFactoriesCache");
+ factories.put("test-controller", factory);
+
Mockito.when(factory.createAccessController(ImmutableMap.of())).thenReturn(registeredAccessController);
+ Mockito.doThrow(new RuntimeException("plugin cleanup
failure")).when(registeredAccessController).close();
+ ExternalCatalog catalog = mockCatalog("test_catalog", 1L);
+
+ withCurrentCatalog(catalog, () ->
accessControllerManager.createAccessController(
+ catalog, "test-controller", ImmutableMap.of(), false));
+ accessControllerManager.removeAccessController("test_catalog",
catalog.getId());
+
+ Mockito.verify(registeredAccessController).close();
+
Assert.assertFalse(accessControllerManager.checkIfAccessControllerExist("test_catalog"));
+ }
+
+ @Test
+ public void testOldGenerationCannotRemoveReplacementController() {
+ CatalogAccessController defaultAccessController =
Mockito.mock(CatalogAccessController.class);
+ CatalogAccessController oldController =
Mockito.mock(CatalogAccessController.class);
+ CatalogAccessController newController =
Mockito.mock(CatalogAccessController.class);
+ AccessControllerFactory factory =
Mockito.mock(AccessControllerFactory.class);
+ AccessControllerManager manager =
createAccessControllerManager(defaultAccessController);
+ ConcurrentHashMap<String, AccessControllerFactory> factories =
+ Deencapsulation.getField(manager,
"accessControllerFactoriesCache");
+ factories.put("test-controller", factory);
+
Mockito.when(factory.createAccessController(ImmutableMap.of())).thenReturn(oldController,
newController);
+ ExternalCatalog oldCatalog = mockCatalog("same_name", 10L);
+ ExternalCatalog newCatalog = mockCatalog("same_name", 11L);
+
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ AtomicReference<CatalogIf> currentCatalog = new
AtomicReference<>(oldCatalog);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ Env env = Mockito.mock(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+
Mockito.when(catalogMgr.getCatalog("same_name")).thenAnswer(invocation ->
currentCatalog.get());
+
+ manager.createAccessController(oldCatalog, "test-controller",
ImmutableMap.of(), false);
+ currentCatalog.set(newCatalog);
+ manager.createAccessController(newCatalog, "test-controller",
ImmutableMap.of(), false);
+ manager.removeAccessController("same_name", oldCatalog.getId());
+
+ Assert.assertSame(newController,
manager.getAccessControllerOrDefault("same_name"));
+ }
+
+ Mockito.verify(oldController).close();
+ Mockito.verify(newController, Mockito.never()).close();
+ }
+
+ @Test
+ public void testControllerPublishedAfterDropIsClosedInsteadOfCached() {
+ CatalogAccessController defaultAccessController =
Mockito.mock(CatalogAccessController.class);
+ CatalogAccessController orphanController =
Mockito.mock(CatalogAccessController.class);
+ AccessControllerFactory factory =
Mockito.mock(AccessControllerFactory.class);
+ AccessControllerManager manager =
createAccessControllerManager(defaultAccessController);
+ ConcurrentHashMap<String, AccessControllerFactory> factories =
+ Deencapsulation.getField(manager,
"accessControllerFactoriesCache");
+ factories.put("test-controller", factory);
+
Mockito.when(factory.createAccessController(ImmutableMap.of())).thenReturn(orphanController);
+ ExternalCatalog droppedCatalog = mockCatalog("dropped", 20L);
+
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ Env env = Mockito.mock(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+ Mockito.when(catalogMgr.getCatalog("dropped")).thenReturn(null);
+
+ manager.createAccessController(droppedCatalog, "test-controller",
ImmutableMap.of(), false);
+ }
+
+ Mockito.verify(orphanController).close();
+ Assert.assertFalse(manager.checkIfAccessControllerExist("dropped"));
+ }
+
+ @Test
+ public void testRemovingFallbackAliasDoesNotCloseSharedDefaultController()
{
+ CatalogAccessController defaultAccessController =
Mockito.mock(CatalogAccessController.class);
+ AccessControllerManager manager =
createAccessControllerManager(defaultAccessController);
+ ExternalCatalog catalog = mockCatalog("fallback", 30L);
+
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ Env env = Mockito.mock(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+
Mockito.when(catalogMgr.getCatalog("fallback")).thenReturn(catalog);
+
+ Assert.assertSame(defaultAccessController,
manager.getAccessControllerOrDefault("fallback"));
+ manager.removeAccessController("fallback", catalog.getId());
+ }
+
+ Mockito.verify(defaultAccessController, Mockito.never()).close();
+ }
+
+ private ExternalCatalog mockCatalog(String name, long id) {
+ ExternalCatalog catalog = Mockito.mock(ExternalCatalog.class);
+ Mockito.when(catalog.getName()).thenReturn(name);
+ Mockito.when(catalog.getId()).thenReturn(id);
+ return catalog;
+ }
+
+ private void withCurrentCatalog(ExternalCatalog catalog, Runnable action) {
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ Env env = Mockito.mock(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+
Mockito.when(catalogMgr.getCatalog(catalog.getName())).thenReturn(catalog);
+ action.run();
+ }
+ }
+
private AccessControllerManager
createAccessControllerManager(CatalogAccessController defaultAccessController) {
AccessControllerManager accessControllerManager = new
AccessControllerManager(new Auth());
Deencapsulation.setField(accessControllerManager,
"defaultAccessController", defaultAccessController);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
