diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java index 9b057a0fbc71af..5c06c726c41890 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java @@ -38,15 +38,15 @@ import java.util.List; import java.util.Optional; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.FutureTask; public class RemoteDorisExternalTable extends ExternalTable { private static final Logger LOG = LogManager.getLogger(RemoteDorisExternalTable.class); private volatile List partitions = Lists.newArrayList(); private volatile List tempPartitions = Lists.newArrayList(); private volatile long tableId = -1; - private volatile boolean isSyncOlapTable = false; - private volatile RemoteOlapTable remoteOlapTable = null; - private volatile Exception lastException = null; + private transient FutureTask currentRefreshTask; public RemoteDorisExternalTable(long id, String name, String remoteName, RemoteDorisExternalCatalog catalog, ExternalDatabase db) { @@ -62,62 +62,58 @@ protected synchronized void makeSureInitialized() { } private RemoteOlapTable getDorisOlapTable() { - if (!isSyncOlapTable) { - synchronized (this) { - if (!isSyncOlapTable) { - try { - isSyncOlapTable = true; - remoteOlapTable = null; - lastException = null; // clear previous exception - - List cachedPartitions = Lists.newArrayList(partitions); - List cachedTempPartitions = Lists.newArrayList(tempPartitions); - RemoteOlapTable olapTable = ((RemoteDorisExternalCatalog) catalog).getFeServiceClient() - .getOlapTable(dbName, remoteName, tableId, cachedPartitions, cachedTempPartitions); - olapTable.setCatalog((RemoteDorisExternalCatalog) catalog); - olapTable.setDatabase((RemoteDorisExternalDatabase) db); - - // Remove redundant nested synchronized block - tableId = olapTable.getId(); - partitions = Lists.newArrayList(olapTable.getPartitions()); - tempPartitions = Lists.newArrayList(olapTable.getTempPartitions().getPartitions()); - - olapTable.setId(id); // change id in case of possible conflicts - olapTable.invalidateBackendsIfNeed(); - remoteOlapTable = olapTable; - } catch (Exception e) { - // Save exception for waiting threads - lastException = e; - LOG.warn("Failed to get remote doris olap table: {}.{}", dbName, remoteName, e); - throw e; // Re-throw the exception - } finally { - isSyncOlapTable = false; - this.notifyAll(); - } - return remoteOlapTable; - } + FutureTask refreshTask; + boolean shouldRun; + synchronized (this) { + if (currentRefreshTask == null || currentRefreshTask.isDone()) { + currentRefreshTask = new FutureTask<>(this::loadDorisOlapTable); + shouldRun = true; + } else { + shouldRun = false; } + refreshTask = currentRefreshTask; } - synchronized (this) { - while (isSyncOlapTable) { - try { - this.wait(); - } catch (InterruptedException e) { - throw new AnalysisException("interrupted while getting doris olap table", e); - } - } + if (shouldRun) { + refreshTask.run(); + } + return getRefreshResult(refreshTask); + } - // If there is a saved exception, throw it with more details - if (remoteOlapTable == null) { - if (lastException != null) { - throw new AnalysisException( - "failed to get remote doris olap table: " + Util.getRootCauseMessage(lastException), - lastException); - } - throw new AnalysisException("failed to get remote doris olap table"); - } - return remoteOlapTable; + private RemoteOlapTable loadDorisOlapTable() { + try { + List cachedPartitions = Lists.newArrayList(partitions); + List cachedTempPartitions = Lists.newArrayList(tempPartitions); + RemoteOlapTable olapTable = ((RemoteDorisExternalCatalog) catalog).getFeServiceClient() + .getOlapTable(dbName, remoteName, tableId, cachedPartitions, cachedTempPartitions); + olapTable.setCatalog((RemoteDorisExternalCatalog) catalog); + olapTable.setDatabase((RemoteDorisExternalDatabase) db); + + tableId = olapTable.getId(); + partitions = Lists.newArrayList(olapTable.getPartitions()); + tempPartitions = Lists.newArrayList(olapTable.getTempPartitions().getPartitions()); + + olapTable.setId(id); // change id in case of possible conflicts + olapTable.invalidateBackendsIfNeed(); + return olapTable; + } catch (RuntimeException e) { + LOG.warn("Failed to get remote doris olap table: {}.{}", dbName, remoteName, e); + throw e; + } + } + + private RemoteOlapTable getRefreshResult(FutureTask refreshTask) { + try { + // The underlying Thrift RPC has its own timeout; this only waits for the shared refresh result. + return refreshTask.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AnalysisException("interrupted while getting doris olap table", e); + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + throw new AnalysisException( + "failed to get remote doris olap table: " + Util.getRootCauseMessage(cause), + cause); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisExternalTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisExternalTableTest.java new file mode 100644 index 00000000000000..7bcfbd5345dc7a --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisExternalTableTest.java @@ -0,0 +1,166 @@ +// 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.doris; + +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.TempPartitions; +import org.apache.doris.nereids.exceptions.AnalysisException; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentMatchers; +import org.mockito.Mockito; +import org.mockito.stubbing.OngoingStubbing; + +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +public class RemoteDorisExternalTableTest { + private static final String DB_NAME = "test_db"; + private static final String REMOTE_TABLE_NAME = "remote_test_table"; + + private RemoteDorisExternalTable table; + private FeServiceClient client; + private RemoteOlapTable remoteOlapTable; + + @BeforeEach + public void setUp() throws Exception { + RemoteDorisExternalCatalog catalog = Mockito.mock(RemoteDorisExternalCatalog.class); + RemoteDorisExternalDatabase db = Mockito.mock(RemoteDorisExternalDatabase.class); + client = Mockito.mock(FeServiceClient.class); + remoteOlapTable = Mockito.mock(RemoteOlapTable.class); + TempPartitions tempPartitions = Mockito.mock(TempPartitions.class); + + Mockito.when(catalog.getId()).thenReturn(1L); + Mockito.doReturn(db).when(catalog).getDbOrAnalysisException(DB_NAME); + Mockito.when(catalog.getFeServiceClient()).thenReturn(client); + Mockito.when(db.getId()).thenReturn(2L); + Mockito.when(db.getFullName()).thenReturn(DB_NAME); + Mockito.when(db.getRemoteName()).thenReturn(DB_NAME); + Mockito.when(remoteOlapTable.getId()).thenReturn(3L); + Mockito.when(remoteOlapTable.getPartitions()).thenReturn(Collections.emptyList()); + Mockito.when(remoteOlapTable.getTempPartitions()).thenReturn(tempPartitions); + Mockito.when(tempPartitions.getPartitions()).thenReturn(Collections.emptyList()); + + table = new RemoteDorisExternalTable( + 4L, "test_table", REMOTE_TABLE_NAME, catalog, db); + } + + @Test + public void testConcurrentRefreshSharesInFlightTask() throws Exception { + CountDownLatch rpcStarted = new CountDownLatch(1); + CountDownLatch releaseRpc = new CountDownLatch(1); + whenRefreshCalled() + .thenAnswer(invocation -> { + rpcStarted.countDown(); + await(releaseRpc); + return remoteOlapTable; + }) + .thenReturn(remoteOlapTable); + + AtomicReference ownerResult = new AtomicReference<>(); + AtomicReference waiterResult = new AtomicReference<>(); + AtomicReference ownerFailure = new AtomicReference<>(); + AtomicReference waiterFailure = new AtomicReference<>(); + + Thread owner = startRefresh(ownerResult, ownerFailure); + Assertions.assertTrue(rpcStarted.await(5, TimeUnit.SECONDS)); + Thread waiter = startRefresh(waiterResult, waiterFailure); + waitUntilWaiting(waiter); + + releaseRpc.countDown(); + join(owner); + join(waiter); + + Assertions.assertNull(ownerFailure.get()); + Assertions.assertNull(waiterFailure.get()); + Assertions.assertSame(remoteOlapTable, ownerResult.get()); + Assertions.assertSame(remoteOlapTable, waiterResult.get()); + verifyRefreshCount(1); + + Assertions.assertSame(remoteOlapTable, table.getOlapTable()); + verifyRefreshCount(2); + } + + @Test + public void testFailedRefreshCanRetry() { + RuntimeException failure = new RuntimeException("refresh failed"); + whenRefreshCalled().thenThrow(failure).thenReturn(remoteOlapTable); + + AnalysisException exception = + Assertions.assertThrows(AnalysisException.class, table::getOlapTable); + Assertions.assertSame(failure, exception.getCause()); + + Assertions.assertSame(remoteOlapTable, table.getOlapTable()); + verifyRefreshCount(2); + } + + private OngoingStubbing whenRefreshCalled() { + return Mockito.when(client.getOlapTable( + ArgumentMatchers.eq(DB_NAME), ArgumentMatchers.eq(REMOTE_TABLE_NAME), + ArgumentMatchers.anyLong(), ArgumentMatchers.anyList(), ArgumentMatchers.anyList())); + } + + private void verifyRefreshCount(int count) { + Mockito.verify(client, Mockito.times(count)).getOlapTable( + ArgumentMatchers.eq(DB_NAME), ArgumentMatchers.eq(REMOTE_TABLE_NAME), + ArgumentMatchers.anyLong(), ArgumentMatchers.anyList(), ArgumentMatchers.anyList()); + } + + private Thread startRefresh(AtomicReference result, + AtomicReference failure) { + Thread thread = new Thread(() -> { + try { + result.set(table.getOlapTable()); + } catch (Throwable t) { + failure.set(t); + } + }); + thread.start(); + return thread; + } + + private static void await(CountDownLatch latch) throws InterruptedException { + if (!latch.await(5, TimeUnit.SECONDS)) { + throw new AssertionError("timed out waiting for test latch"); + } + } + + private static void waitUntilWaiting(Thread thread) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (System.nanoTime() < deadline) { + Thread.State state = thread.getState(); + if (state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING) { + return; + } + if (!thread.isAlive()) { + throw new AssertionError("thread exited before waiting"); + } + Thread.sleep(10); + } + throw new AssertionError("thread did not enter waiting state"); + } + + private static void join(Thread thread) throws InterruptedException { + thread.join(TimeUnit.SECONDS.toMillis(5)); + Assertions.assertFalse(thread.isAlive(), "test thread did not finish"); + } +}