This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a3330994747 Fix scan termination for finished fragment instances
(#18701)
a3330994747 is described below
commit a333099474732e557ca373d6f096ddc7d8aea08d
Author: Jackie Tien <[email protected]>
AuthorDate: Tue Sep 22 19:27:53 2026 +0800
Fix scan termination for finished fragment instances (#18701)
---
.../db/queryengine/execution/driver/Driver.java | 6 +
.../FragmentInstanceFinishedException.java | 38 +++++
.../execution/operator/source/SeriesScanUtil.java | 4 +
.../source/SeriesScanUtilCancellationTest.java | 172 ++++++++++++++++++++-
.../schedule/DefaultDriverSchedulerTest.java | 81 ++++++++++
5 files changed, 294 insertions(+), 7 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/Driver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/Driver.java
index 3d76cfd84f4..8f3c02600bf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/Driver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/Driver.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import org.apache.iotdb.db.queryengine.execution.exchange.sink.ISink;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceFinishedException;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.schedule.task.DriverTaskId;
import org.apache.iotdb.db.queryengine.metric.QueryMetricsManager;
@@ -248,6 +249,11 @@ public abstract class Driver implements IDriver {
}
}
return NOT_BLOCKED;
+ } catch (FragmentInstanceFinishedException e) {
+ // The fragment may finish before its notification thread closes this
driver. Release the
+ // driver through the normal cleanup path without aborting other
fragments of the query.
+ state.compareAndSet(State.ALIVE, State.NEED_DESTRUCTION);
+ return NOT_BLOCKED;
} catch (Throwable t) {
Throwable actualCause = t;
if (actualCause.getCause() instanceof IoTDBRuntimeException) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceFinishedException.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceFinishedException.java
new file mode 100644
index 00000000000..c896883fae7
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceFinishedException.java
@@ -0,0 +1,38 @@
+/*
+ * 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.iotdb.db.queryengine.execution.fragment;
+
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
+import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
+
+/**
+ * Internal control signal for stopping a scan after its fragment has
finished. This is unchecked so
+ * scan operators do not wrap it as an I/O failure before the driver can
handle normal termination.
+ */
+public class FragmentInstanceFinishedException extends RuntimeException {
+
+ public FragmentInstanceFinishedException(FragmentInstanceId
fragmentInstanceId) {
+ super(
+ String.format(
+
DataNodeQueryMessages.EXCEPTION_FRAGMENT_INSTANCE_ARG_IS_ALREADY_ARG_B44984B4,
+ fragmentInstanceId,
+ FragmentInstanceState.FINISHED));
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
index c00f092c4b5..1582cacabc6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.path.NonAlignedFullPath;
import org.apache.iotdb.db.exception.CorruptedTsFileException;
import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceFinishedException;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceState;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet;
@@ -1587,6 +1588,9 @@ public class SeriesScanUtil implements Accountable {
return;
}
FragmentInstanceState state = context.getStateMachine().getState();
+ if (state == FragmentInstanceState.FINISHED) {
+ throw new FragmentInstanceFinishedException(context.getId());
+ }
if (state.isDone()) {
// A scan over many overlapping files may stay in one operator call long
after cancellation.
// Exit on the driver thread so it can release its lock and finish
resource cleanup.
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
index a32a83e1e02..e347699cb6d 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
@@ -22,6 +22,7 @@ package
org.apache.iotdb.db.queryengine.execution.operator.source;
import org.apache.iotdb.calc.execution.operator.Operator;
import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager;
import org.apache.iotdb.commons.exception.QueryTimeoutException;
+import org.apache.iotdb.commons.exception.SemanticException;
import org.apache.iotdb.commons.path.AlignedFullPath;
import org.apache.iotdb.commons.path.NonAlignedFullPath;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -34,6 +35,7 @@ import
org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeManager
import org.apache.iotdb.db.queryengine.execution.exchange.sink.ISink;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceExecution;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceFinishedException;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceState;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.queryengine.execution.schedule.IDriverScheduler;
@@ -63,6 +65,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.IntConsumer;
@@ -71,9 +74,11 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -139,8 +144,12 @@ public class SeriesScanUtilCancellationTest {
throw new AssertionError(state);
}
- IOException exception = assertThrows(IOException.class,
scanner::hasNextFile);
- assertSame(context.getFailureCause().orElse(null), exception.getCause());
+ if (state == FragmentInstanceState.FINISHED) {
+ assertThrows(FragmentInstanceFinishedException.class,
scanner::hasNextFile);
+ } else {
+ IOException exception = assertThrows(IOException.class,
scanner::hasNextFile);
+ assertSame(context.getFailureCause().orElse(null),
exception.getCause());
+ }
assertEquals(0, loadedFiles);
assertEquals(state, context.getStateMachine().getState());
}
@@ -237,12 +246,22 @@ public class SeriesScanUtilCancellationTest {
@Test(timeout = 15000)
public void testTimeoutUnblocksDriverResourceCleanup() throws Exception {
+ assertFailureUnblocksDriverResourceCleanup(new QueryTimeoutException());
+ }
+
+ @Test(timeout = 15000)
+ public void testSemanticFailureUnblocksDriverResourceCleanup() throws
Exception {
+ assertFailureUnblocksDriverResourceCleanup(
+ new SemanticException("Scalar sub-query has returned multiple rows."));
+ }
+
+ private void assertFailureUnblocksDriverResourceCleanup(RuntimeException
failure)
+ throws Exception {
ExecutorService notifications = Executors.newSingleThreadExecutor();
FragmentInstanceContext context = newContext(notifications);
context.initializeNumOfDrivers(1);
try {
CountDownLatch closeRequested = new CountDownLatch(1);
- QueryTimeoutException timeout = new QueryTimeoutException();
SeriesScanUtil scanner =
newScanner(
context,
@@ -251,7 +270,7 @@ public class SeriesScanUtilCancellationTest {
4,
count -> {
if (count == 2) {
- context.failed(timeout);
+ context.failed(failure);
try {
// The notification thread has requested close while this
thread owns the
// driver lock, just as in a query that is still loading
metadata.
@@ -296,10 +315,9 @@ public class SeriesScanUtilCancellationTest {
exchangeManager);
assertSame(
- timeout,
+ failure,
assertThrows(
- QueryTimeoutException.class,
- () -> driver.processFor(new Duration(1, TimeUnit.SECONDS))));
+ RuntimeException.class, () -> driver.processFor(new Duration(1,
TimeUnit.SECONDS))));
// This can complete only after the preceding cleanup callback gets past
allDriversClosed.
notifications.submit(() -> {}).get(5, TimeUnit.SECONDS);
assertEquals(2, loadedFiles);
@@ -316,6 +334,146 @@ public class SeriesScanUtilCancellationTest {
}
}
+ @Test(timeout = 60000)
+ public void testFinishedScanUnblocksDriverResourceCleanup() throws Exception
{
+ for (boolean sequence : new boolean[] {true, false}) {
+ for (boolean waitForClose : new boolean[] {false, true}) {
+ assertFinishedScanUnblocksDriverResourceCleanup(sequence,
waitForClose);
+ }
+ }
+ }
+
+ private void assertFinishedScanUnblocksDriverResourceCleanup(
+ boolean sequence, boolean waitForClose) throws Exception {
+ ExecutorService notifications = Executors.newSingleThreadExecutor();
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ CountDownLatch allowNotifications = new CountDownLatch(waitForClose ? 0 :
1);
+ CountDownLatch scanEntered = new CountDownLatch(1);
+ CountDownLatch resumeScan = new CountDownLatch(1);
+ CountDownLatch closeRequested = new CountDownLatch(1);
+ // Cover both orderings: the FI is FINISHED before driver.close(), and
close is already pending.
+ notifications.submit(() -> await(allowNotifications));
+ FragmentInstanceContext context = newContext(notifications);
+ context.initializeNumOfDrivers(1);
+ Operator operator = mock(Operator.class);
+ doReturn(NOT_BLOCKED).when(operator).isBlocked();
+ when(operator.hasNextWithTimer()).thenReturn(true);
+ SeriesScanUtil scanner =
+ newScanner(
+ context,
+ Ordering.ASC,
+ sequence,
+ 4,
+ count -> {
+ if (count == 1) {
+ scanEntered.countDown();
+ await(resumeScan);
+ }
+ });
+ when(operator.nextWithTimer())
+ .thenAnswer(
+ invocation -> {
+ scanner.hasNextFile();
+ return null;
+ });
+ ISink sink = mock(ISink.class);
+ doReturn(NOT_BLOCKED).when(sink).isFull();
+ DataDriverContext driverContext = new DataDriverContext(context, 0);
+ driverContext.setSink(sink);
+ DataDriver driver =
+ new DataDriver(operator, driverContext, 0) {
+ @Override
+ public void close() {
+ super.close();
+ closeRequested.countDown();
+ }
+ };
+ MPPDataExchangeManager exchangeManager =
mock(MPPDataExchangeManager.class);
+ IDriverScheduler scheduler = mock(IDriverScheduler.class);
+ FragmentInstanceExecution.createFragmentInstanceExecution(
+ scheduler,
+ context.getId(),
+ context,
+ Collections.singletonList(driver),
+ sink,
+ context.getStateMachine(),
+ 1000,
+ false,
+ exchangeManager);
+ try {
+ Future<?> execution =
+ worker.submit(() -> driver.processFor(new Duration(1,
TimeUnit.SECONDS)));
+ await(scanEntered);
+ context.finished();
+ if (waitForClose) {
+ await(closeRequested);
+ }
+ resumeScan.countDown();
+ execution.get(5, TimeUnit.SECONDS);
+ assertTrue(driver.isFinished());
+ assertEquals(1, loadedFiles);
+ assertEquals(FragmentInstanceState.FINISHED,
context.getStateMachine().getState());
+ assertTrue(context.getStateMachine().getFailureCauses().isEmpty());
+ allowNotifications.countDown();
+ notifications.submit(() -> {}).get(5, TimeUnit.SECONDS);
+ verify(operator).close();
+ verify(context.getMemoryReservationContext()).releaseAllReservedMemory();
+ verify(exchangeManager)
+ .deRegisterFragmentInstanceFromMemoryPool(
+ context.getId().getQueryId().getId(),
context.getId().getFragmentInstanceId(), true);
+ verify(scheduler, never()).abortFragmentInstance(any(), any());
+ } finally {
+ resumeScan.countDown();
+ allowNotifications.countDown();
+ worker.shutdownNow();
+ assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
+ driver.close();
+ context.decrementNumOfUnClosedDriver();
+ // Let FI cleanup finish after opening its latch, without interrupting
its driver-close wait.
+ notifications.shutdown();
+ assertTrue(notifications.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ public void testFinishedFragmentDoesNotSuppressIOException() throws
Exception {
+ FragmentInstanceContext context = newContext(Runnable::run);
+ context.initializeNumOfDrivers(1);
+ IOException failure = new IOException("Failed to read file metadata");
+ Operator operator = mock(Operator.class);
+ doReturn(NOT_BLOCKED).when(operator).isBlocked();
+ when(operator.hasNextWithTimer()).thenReturn(true);
+ when(operator.nextWithTimer())
+ .thenAnswer(
+ invocation -> {
+ context.finished();
+ throw failure;
+ });
+ ISink sink = mock(ISink.class);
+ doReturn(NOT_BLOCKED).when(sink).isFull();
+ DataDriverContext driverContext = new DataDriverContext(context, 0);
+ driverContext.setSink(sink);
+ DataDriver driver = new DataDriver(operator, driverContext, 0);
+ try {
+ RuntimeException exception =
+ assertThrows(
+ RuntimeException.class, () -> driver.processFor(new Duration(1,
TimeUnit.SECONDS)));
+ assertSame(failure, exception.getCause());
+ assertSame(failure, context.getFailureCause().get());
+ } finally {
+ driver.close();
+ }
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ assertTrue(latch.await(5, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ }
+
private FragmentInstanceContext newContext(Executor executor) {
FragmentInstanceId id =
new FragmentInstanceId(new PlanFragmentId(new
QueryId("scan_cancellation"), 0), "0");
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/DefaultDriverSchedulerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/DefaultDriverSchedulerTest.java
index 9b3cb38cbc0..4f9814b9ed6 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/DefaultDriverSchedulerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/DefaultDriverSchedulerTest.java
@@ -18,14 +18,21 @@
*/
package org.apache.iotdb.db.queryengine.execution.schedule;
+import org.apache.iotdb.calc.execution.operator.Operator;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.execution.driver.DataDriver;
+import org.apache.iotdb.db.queryengine.execution.driver.DataDriverContext;
import org.apache.iotdb.db.queryengine.execution.driver.DriverContext;
import org.apache.iotdb.db.queryengine.execution.driver.IDriver;
import
org.apache.iotdb.db.queryengine.execution.exchange.IMPPDataExchangeManager;
+import org.apache.iotdb.db.queryengine.execution.exchange.sink.ISink;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceFinishedException;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceState;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine;
import
org.apache.iotdb.db.queryengine.execution.schedule.queue.multilevelqueue.DriverTaskHandle;
import
org.apache.iotdb.db.queryengine.execution.schedule.queue.multilevelqueue.MultilevelPriorityQueue;
@@ -50,6 +57,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
+import static org.apache.iotdb.calc.execution.operator.Operator.NOT_BLOCKED;
import static
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
public class DefaultDriverSchedulerTest {
@@ -530,6 +538,79 @@ public class DefaultDriverSchedulerTest {
}
}
+ @Test
+ public void testFinishedFragmentDoesNotAbortOtherTasks() throws Exception {
+ int previousDataNodeId =
IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+ IoTDBDescriptor.getInstance().getConfig().setDataNodeId(1);
+ IMPPDataExchangeManager exchangeManager =
Mockito.mock(IMPPDataExchangeManager.class);
+ manager.setBlockManager(exchangeManager);
+ QueryId queryId = new QueryId("finished_scan");
+ FragmentInstanceId finishedId = new FragmentInstanceId(new
PlanFragmentId(queryId, 0), "0");
+ FragmentInstanceContext context =
+ createFragmentInstanceContext(
+ finishedId, new FragmentInstanceStateMachine(finishedId,
Runnable::run));
+ context.initializeNumOfDrivers(1);
+ Operator operator = Mockito.mock(Operator.class);
+ Mockito.doReturn(NOT_BLOCKED).when(operator).isBlocked();
+ Mockito.when(operator.hasNextWithTimer()).thenReturn(true);
+ Mockito.when(operator.nextWithTimer())
+ .thenAnswer(
+ invocation -> {
+ context.finished();
+ throw new FragmentInstanceFinishedException(finishedId);
+ });
+ ISink sink = Mockito.mock(ISink.class);
+ Mockito.doReturn(NOT_BLOCKED).when(sink).isFull();
+ DataDriverContext driverContext = new DataDriverContext(context, 0);
+ driverContext.setSink(sink);
+ DataDriver driver = new DataDriver(operator, driverContext, 0);
+ try {
+ DriverTaskHandle handle =
+ new DriverTaskHandle(
+ 1,
+ (MultilevelPriorityQueue) manager.getReadyQueue(),
+ OptionalInt.of(Integer.MAX_VALUE));
+ DriverTask task = new DriverTask(driver, 30000, DriverTaskStatus.READY,
handle, 0, false);
+ manager.registerTaskToQueryMap(queryId, task);
+ manager.getTimeoutQueue().push(task);
+ manager.submitTaskToReadyQueue(task);
+ Assert.assertSame(task, manager.getReadyQueue().poll());
+
+ FragmentInstanceId siblingId = new FragmentInstanceId(new
PlanFragmentId(queryId, 1), "0");
+ IDriver siblingDriver = Mockito.mock(IDriver.class);
+ Mockito.when(siblingDriver.getDriverTaskId()).thenReturn(new
DriverTaskId(siblingId, 0));
+ DriverTask sibling =
+ new DriverTask(siblingDriver, 30000, DriverTaskStatus.READY, handle,
0, false);
+ manager.registerTaskToQueryMap(queryId, sibling);
+ manager.getTimeoutQueue().push(sibling);
+ manager.submitTaskToReadyQueue(sibling);
+
+ DriverTaskThread worker =
+ new DriverTaskThread("finished-scan", null, null,
manager.getScheduler(), null);
+ worker.execute(task);
+
+ Assert.assertEquals(FragmentInstanceState.FINISHED,
context.getStateMachine().getState());
+
Assert.assertTrue(context.getStateMachine().getFailureCauses().isEmpty());
+ Assert.assertEquals(DriverTaskStatus.FINISHED, task.getStatus());
+ Assert.assertFalse(task.getAbortCause().isPresent());
+ Assert.assertNull(manager.getTimeoutQueue().get(task.getDriverTaskId()));
+
Assert.assertFalse(manager.getQueryMap().get(queryId).containsKey(finishedId));
+ Assert.assertEquals(DriverTaskStatus.READY, sibling.getStatus());
+ Assert.assertFalse(sibling.getAbortCause().isPresent());
+ Assert.assertEquals(1, manager.getReadyQueueTaskCount());
+ Assert.assertSame(sibling, manager.getReadyQueue().poll());
+ Assert.assertSame(sibling,
manager.getTimeoutQueue().get(sibling.getDriverTaskId()));
+
Assert.assertTrue(manager.getQueryMap().get(queryId).get(siblingId).contains(sibling));
+ Mockito.verify(operator).close();
+ Mockito.verify(siblingDriver, Mockito.never()).failed(Mockito.any());
+ Mockito.verify(exchangeManager, Mockito.never())
+ .forceDeregisterFragmentInstance(Mockito.any());
+ } finally {
+ driver.close();
+
IoTDBDescriptor.getInstance().getConfig().setDataNodeId(previousDataNodeId);
+ }
+ }
+
private void clear() {
manager.getQueryMap().clear();
manager.getBlockedTasks().clear();
