apache · 924060929 · Aug 19, 2026 · Aug 19, 2026 · Aug 19, 2026 · Aug 20, 2026
@@ -36,6 +36,7 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.FutureTask;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.PriorityBlockingQueue;
@@ -390,6 +391,7 @@ public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
     public static class BlockedPolicy implements RejectedExecutionHandler {
 
         private static final Logger LOG = LogManager.getLogger(BlockedPolicy.class);
+        private static final long CANCEL_CHECK_INTERVAL_NANOS = TimeUnit.MILLISECONDS.toNanos(100);
 
         private String threadPoolName;
 
@@ -403,18 +405,34 @@ public BlockedPolicy(String threadPoolName, int timeoutSeconds) {
         @Override
         public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
             try {
-                boolean ret = executor.getQueue().offer(r, timeoutSeconds, TimeUnit.SECONDS);
-                if (!ret) {
-                    throw new RejectedExecutionException("submit task failed, queue size is full: "
-                            + this.threadPoolName);
+                if (isCancelled(r) || executor.getQueue().offer(r)) {
+                    return;
+                }
+                long remainingNanos = TimeUnit.SECONDS.toNanos(timeoutSeconds);
+                while (!isCancelled(r)) {
+                    if (remainingNanos <= 0) {
+                        throw new RejectedExecutionException("submit task failed, queue size is full: "
+                                + this.threadPoolName);
+                    }
+                    long waitNanos = Math.min(remainingNanos, CANCEL_CHECK_INTERVAL_NANOS);
+                    long startNanos = System.nanoTime();
+                    if (executor.getQueue().offer(r, waitNanos, TimeUnit.NANOSECONDS)) {
+                        return;
+                    }
+                    remainingNanos -= System.nanoTime() - startNanos;
                 }
             } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
                 String errMsg = String.format("Task %s wait to enqueue in %s %s failed",
                         r.toString(), threadPoolName, executor.toString());
                 LOG.warn(errMsg);
                 throw new RejectedExecutionException(errMsg);
             }
         }
+
+        private boolean isCancelled(Runnable task) {
+            return task instanceof Future<?> && ((Future<?>) task).isCancelled();
+        }
     }
 
     static class LogDiscardOldestPolicy implements RejectedExecutionHandler {

@@ -740,8 +740,6 @@ private void alterExternalCatalogPropsFenced(ExternalCatalog externalCatalog, Ca
                 throw new DdlException("Invalid catalog properties: "
                         + validationException.getMessage(), validationException);
             }
-        } else {
-            externalCatalog.tryModifyCatalogProps(newProps);
         }
         if (newProps.containsKey(METADATA_REFRESH_INTERVAL_SEC)) {
             long catalogId = externalCatalog.getId();

@@ -191,11 +191,16 @@ public void finishSchedule() {
     }
 
     public void stop() {
-        if (isStop()) {
-            return;
+        List<Closeable> resources;
+        synchronized (closeableResources) {
+            if (isStop()) {
+                return;
+            }
+            isStopped.set(true);
+            resources = new ArrayList<>(closeableResources);
+            closeableResources.clear();
         }
-        isStopped.set(true);
-        closeableResources.forEach((closeable) -> {
+        resources.forEach((closeable) -> {
             try {
                 closeable.close();
             } catch (Exception e) {
@@ -204,16 +209,23 @@ public void stop() {
             }
         });
         notifyAssignment();
-        if (exception != null) {
-            throw new RuntimeException(exception);
-        }
     }
 
     public boolean isStop() {
         return isStopped.get();
     }
 
     public void addCloseable(Closeable resource) {
-        closeableResources.add(resource);
+        synchronized (closeableResources) {
+            if (!isStop()) {
+                closeableResources.add(resource);
+                return;
+            }
+        }
+        try {
+            resource.close();
+        } catch (Exception e) {
+            LOG.warn("close resource registered after stop error:{}", e.getMessage(), e);
+        }
     }
 }