github-actions[bot] commented on code in PR #66913: URL: https://github.com/apache/doris/pull/66913#discussion_r4004865868
########## fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogRuntimeStateTest.java: ########## @@ -0,0 +1,120 @@ +// 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.datasource.hive.HMSExternalCatalog; +import org.apache.doris.datasource.iceberg.IcebergCatalogResourceTracker; +import org.apache.doris.datasource.iceberg.IcebergExternalCatalog; +import org.apache.doris.datasource.iceberg.IcebergRestExternalCatalog; +import org.apache.doris.persist.gson.GsonUtils; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.HashMap; +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.TimeoutException; + +public class ExternalCatalogRuntimeStateTest { + + @Test + public void testIcebergRuntimeStateRestoredAfterGsonReplay() throws Exception { + IcebergExternalCatalog restored = roundTrip( + new IcebergRestExternalCatalog(1L, "iceberg", null, Collections.emptyMap(), ""), + IcebergExternalCatalog.class); + + assertTrackerCanRetainAndRelease(restored, IcebergExternalCatalog.class, "resourceTracker"); + } + + @Test + public void testHmsRuntimeStateRestoredAfterGsonReplay() throws Exception { + HMSExternalCatalog restored = roundTrip( + new HMSExternalCatalog(2L, "hms", null, Collections.emptyMap(), ""), + HMSExternalCatalog.class); + + Assertions.assertEquals(0L, restored.getRuntimeGeneration()); + assertTrackerCanRetainAndRelease(restored, HMSExternalCatalog.class, "icebergResourceTracker"); + } + + @Test + public void testHmsRuntimeGenerationCannotObservePropertyCommitMidReset() throws Exception { + CountDownLatch resetEntered = new CountDownLatch(1); + CountDownLatch allowResetToFinish = new CountDownLatch(1); + HMSExternalCatalog catalog = new HMSExternalCatalog( + 3L, "hms", null, + new HashMap<>(Collections.singletonMap("s3.access_key", "old")), "") { + @Override + public synchronized void notifyPropertiesUpdated(java.util.Map<String, String> updatedProps) { + resetEntered.countDown(); + try { + Assertions.assertTrue(allowResetToFinish.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + }; + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future<?> modifier = executor.submit(() -> + catalog.modifyCatalogProps(Collections.singletonMap("s3.access_key", "new"))); + Assertions.assertTrue(resetEntered.await(5, TimeUnit.SECONDS)); + Assertions.assertEquals("new", catalog.getProperties().get("s3.access_key")); + + CountDownLatch readerStarted = new CountDownLatch(1); + Future<Long> reader = executor.submit(() -> { + readerStarted.countDown(); + return catalog.getRuntimeGeneration(); + }); + Assertions.assertTrue(readerStarted.await(5, TimeUnit.SECONDS)); + Assertions.assertThrows(TimeoutException.class, () -> reader.get(200, TimeUnit.MILLISECONDS)); Review Comment: [P1] Update this test for the lock-free generation read The modifier increments `runtimeGeneration` before it enters the paused `notifyPropertiesUpdated` hook. Since `getRuntimeGeneration()` is now an unsynchronized `AtomicLong.get()`, this reader normally completes immediately with 1, so the `TimeoutException` assertion fails (or depends on an unrelated scheduler pause). Remove the stale blocking oracle and assert the immediate incremented generation instead. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/rewrite/RewriteGroupTask.java: ########## @@ -186,6 +212,9 @@ private void executeGroup(ConnectContext taskConnectContext, StatementBase taskParsedStmt) throws Exception { // Step 1: Create stmt executor stmtExecutor = new StmtExecutor(taskConnectContext, taskParsedStmt); + if (isCanceled.get()) { Review Comment: [P1] Publish the rewrite coordinator before relying on cancellation This new check is the last place a rewrite task observes `isCanceled`. After it passes, `cancel()` routes through `stmtExecutor.cancel()`, but this rewrite path never calls `stmtExecutor.setCoord(insertExecutor.getCoordinator())` as normal inserts and the other Iceberg DML commands do. Cancellation of every running group is therefore a no-op: a sibling failure, timeout, interruption, or later submission rejection can roll back the outer transaction while accepted BE rewrites keep consuming resources and writing uncommitted files. Publish the actual coordinator with a register-or-cancel fence, and cover a running group cancelled during failure cleanup. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java: ########## @@ -173,17 +240,108 @@ protected List<String> listTableNamesFromRemote(SessionContext ctx, String dbNam } @Override - public void onClose() { + public synchronized void onClose() { + ThreadPoolExecutor retiredExecutor = threadPoolWithPreAuth; + threadPoolWithPreAuth = null; super.onClose(); - if (null != catalog) { - try { - if (catalog instanceof AutoCloseable) { - ((AutoCloseable) catalog).close(); - } - catalog = null; - } catch (Exception e) { - LOG.warn("Failed to close iceberg catalog: {}", getName(), e); + Catalog retiredCatalog = catalog; + catalog = null; + resourceTracker.retireCurrent(() -> { Review Comment: [P1] Keep catalog teardown out of the cache-removal lock stack `CatalogMgr` calls `onClose()` before `removeCatalog`/`removeCatalogPermanently`, so this retires the tracker while cached table values still own its references. The later group close runs the new synchronous `value.retire()` extractor under the catalog lifecycle stripe, Iceberg cache monitor, admission lock, and Caffeine removal locks; releasing the last value then executes `closeCatalog()` and up to two 60-second executor waits on that stack, blocking unrelated cache work. This is separate from the operation-lease thread on this line: even quiescent cached owners trigger this teardown order. Retire the cache group first, as reset does, or dispatch final teardown after those locks are released. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/HudiExternalMetaCache.java: ########## @@ -93,8 +96,10 @@ public HudiExternalMetaCache(ExecutorService refreshExecutor, ExternalMetaCacheB TablePartitionValues.class, this::loadPartitionValuesCacheValue, defaultEntryCacheSpec(), MetaCacheEntryInvalidation.forNameMapping(HudiPartitionCacheKey::getNameMapping))); fsViewEntry = registerEntry(MetaCacheEntryDef.of(ENTRY_FS_VIEW, HudiFsViewCacheKey.class, - HoodieTableFileSystemView.class, this::createFsView, defaultEntryCacheSpec(), - MetaCacheEntryInvalidation.forNameMapping(HudiFsViewCacheKey::getNameMapping))); + HudiFsViewCacheValue.class, this::createFsView, defaultEntryCacheSpec(), + false, MetaCacheEntryInvalidation.forNameMapping(HudiFsViewCacheKey::getNameMapping)) + .withRemovalListener(value -> value, this::releaseFsViewCacheReference) Review Comment: [P1] Keep retired Hudi views inside a bounded budget This extractor still puts the full `HudiFsViewCacheValue`, including its `HoodieTableFileSystemView` timeline/file-slice graph, onto `MetaCacheEntry`'s unbounded process-wide cleanup queue after the live entry stops counting toward cache capacity. If cleanup is slow, eviction or invalidation churn can admit replacement views while arbitrarily many retired graphs remain reachable. The analogous Iceberg extractor now retires synchronously and queues only a `Boolean`; this Hudi registration still retains the graph. Retire through a bounded/accounted owner or otherwise keep this cleanup backlog inside the cache budget. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
