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 ccbc71d60cb Fix misleading shutdown interruption warnings (#18939)
ccbc71d60cb is described below
commit ccbc71d60cb42f049601cd1eebe72ff0702b8f0b
Author: Jackie Tien <[email protected]>
AuthorDate: Wed Oct 21 10:01:29 2026 +0800
Fix misleading shutdown interruption warnings (#18939)
---
.../iotdb/db/i18n/DataNodeQueryMessages.java | 2 +
.../iotdb/db/i18n/DataNodeQueryMessages.java | 2 +
.../queryengine/execution/QueryStateMachine.java | 49 +-
.../execution/schedule/AbstractDriverThread.java | 16 +-
.../schedule/DriverTaskAbortedException.java | 2 +
.../plan/execution/ExecutionResult.java | 8 +
.../queryengine/plan/execution/QueryExecution.java | 25 +-
.../plan/scheduler/ClusterScheduler.java | 3 +-
.../plan/scheduler/FragInstanceDispatchResult.java | 12 +
.../scheduler/FragmentInstanceDispatcherImpl.java | 6 +-
.../metrics/IoTDBInternalLocalReporter.java | 156 ++++--
.../execution/QueryStateMachineTest.java | 35 ++
.../schedule/AbstractDriverThreadShutdownTest.java | 271 +++++++++
.../plan/relational/analyzer/TreeViewTest.java | 12 +-
.../metrics/IoTDBInternalLocalReporterTest.java | 623 +++++++++++++++++++++
15 files changed, 1164 insertions(+), 58 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index db0fa44dbc5..7dd677cab3f 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -1263,6 +1263,7 @@ public final class DataNodeQueryMessages {
"Interrupted when dispatching read async";
public static final String INTERRUPTED_WHEN_DISPATCHING_WRITE_ASYNC =
"Interrupted when dispatching write async";
+ public static final String MESSAGE_INTERRUPTED_ERRORS_ARG_876F1B4B =
"Interrupted errors: %s";
public static final String DESERIALIZE_CONSENSUSGROUPID_FAILED =
"Deserialize ConsensusGroupId failed. ";
public static final String CAN_T_CONNECT_TO_NODE =
@@ -3476,6 +3477,7 @@ public final class DataNodeQueryMessages {
"already being cancelled";
public static final String DRIVER_TASK_ABORTED_BY_INTERNAL_ERROR_SCHEDULED =
"internal error scheduled";
+ public static final String EXCEPTION_SCHEDULER_SHUTDOWN_323938F5 =
"scheduler shutdown";
public static final String DRIVER_TASK_ABORTED_BY_MEMORY_NOT_ENOUGH =
"Memory is not enough to execute the query task.";
public static final String ROOT_FRAGMENT_INSTANCE_PLACEMENT_ERROR_FMT =
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index f500838c1ca..203c7552ae9 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -1243,6 +1243,7 @@ public final class DataNodeQueryMessages {
"异步分发读取操作时被中断";
public static final String INTERRUPTED_WHEN_DISPATCHING_WRITE_ASYNC =
"异步分发写入操作时被中断";
+ public static final String MESSAGE_INTERRUPTED_ERRORS_ARG_876F1B4B =
"中断错误：%s";
public static final String DESERIALIZE_CONSENSUSGROUPID_FAILED =
"反序列化 ConsensusGroupId 失败。";
public static final String CAN_T_CONNECT_TO_NODE =
@@ -4127,6 +4128,7 @@ public final class DataNodeQueryMessages {
"已经在取消中";
public static final String DRIVER_TASK_ABORTED_BY_INTERNAL_ERROR_SCHEDULED =
"内部错误调度";
+ public static final String EXCEPTION_SCHEDULER_SHUTDOWN_323938F5 = "调度器关闭";
public static final String DRIVER_TASK_ABORTED_BY_MEMORY_NOT_ENOUGH =
"内存不足，无法执行查询任务。";
public static final String ROOT_FRAGMENT_INSTANCE_PLACEMENT_ERROR_FMT =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachine.java
index 1bc3e2b0f32..2683fec32cf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachine.java
@@ -54,7 +54,7 @@ public class QueryStateMachine {
private final StateMachine<QueryState> queryState;
private final AtomicReference<Throwable> failureException = new
AtomicReference<>();
- private final AtomicReference<TSStatus> failureStatus = new
AtomicReference<>();
+ private final AtomicReference<FailureStatus> failureStatus = new
AtomicReference<>();
public QueryStateMachine(QueryId queryId, ExecutorService executor) {
this.queryState =
@@ -90,7 +90,7 @@ public class QueryStateMachine {
}
public void transitionToPendingRetry(TSStatus failureStatus) {
- this.failureStatus.compareAndSet(null, failureStatus);
+ recordFailureStatus(failureStatus, false);
queryState.setIf(PENDING_RETRY, currentState -> currentState ==
DISPATCHING);
}
@@ -110,7 +110,7 @@ public class QueryStateMachine {
}
public boolean transitionToCanceled(Throwable throwable, TSStatus
failureStatus) {
- this.failureStatus.compareAndSet(null, failureStatus);
+ recordFailureStatus(failureStatus, false);
this.failureException.compareAndSet(null, throwable);
return transitionToDoneState(CANCELED);
}
@@ -129,10 +129,34 @@ public class QueryStateMachine {
}
public void transitionToFailed(TSStatus failureStatus) {
- this.failureStatus.compareAndSet(null, failureStatus);
+ transitionToFailed(failureStatus, false);
+ }
+
+ public void transitionToFailed(TSStatus failureStatus, boolean
failureCausedByInterruption) {
+ recordFailureStatus(failureStatus, failureCausedByInterruption);
transitionToDoneState(FAILED);
}
+ private void recordFailureStatus(TSStatus status, boolean
failureCausedByInterruption) {
+ if (status != null) {
+ // Publish the status and its origin together. A later interruption must
not reclassify
+ // the first recorded failure.
+ failureStatus.compareAndSet(null, new FailureStatus(status,
failureCausedByInterruption));
+ }
+ }
+
+ /** Only the exact recorded status carries the interruption provenance
stored with it. */
+ public boolean isFailureCausedByInterruption(TSStatus status) {
+ FailureStatus failure = failureStatus.get();
+ if (failure == null || failure.status != status ||
!failure.failureCausedByInterruption) {
+ return false;
+ }
+ // Statuses and exceptions can be recorded independently. Keep an
unrelated exception
+ // visible even if a subsequent dispatch interruption supplied the status.
+ Throwable throwable = failureException.get();
+ return throwable == null || getRootCause(throwable) instanceof
InterruptedException;
+ }
+
private boolean transitionToDoneState(QueryState doneState) {
requireNonNull(doneState,
DataNodeQueryMessages.EXCEPTION_DONESTATE_IS_NULL_D88F77E5);
checkArgument(
@@ -161,9 +185,9 @@ public class QueryStateMachine {
}
public TSStatus getFailureStatus() {
- TSStatus status = failureStatus.get();
+ FailureStatus status = failureStatus.get();
if (status != null) {
- return status;
+ return status.status;
} else {
Throwable throwable = failureException.get();
if (throwable != null) {
@@ -174,7 +198,18 @@ public class QueryStateMachine {
return RpcUtils.getStatus(((IoTDBException) t).getErrorCode(),
t.getMessage());
}
}
- return failureStatus.get();
+ status = failureStatus.get();
+ return status == null ? null : status.status;
+ }
+ }
+
+ private static class FailureStatus {
+ private final TSStatus status;
+ private final boolean failureCausedByInterruption;
+
+ private FailureStatus(TSStatus status, boolean
failureCausedByInterruption) {
+ this.status = status;
+ this.failureCausedByInterruption = failureCausedByInterruption;
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThread.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThread.java
index ac34cccb437..ee5eb8623bf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThread.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThread.java
@@ -67,8 +67,11 @@ public abstract class AbstractDriverThread extends Thread
implements Closeable {
try {
next = queue.poll();
} catch (InterruptedException e) {
- logger.warn(
-
DataNodeQueryMessages.EXECUTOR_FAILED_TO_POLL_DRIVER_TASK_FROM_QUEUE,
this.getName());
+ if (!closed) {
+ logger.warn(
+
DataNodeQueryMessages.EXECUTOR_FAILED_TO_POLL_DRIVER_TASK_FROM_QUEUE,
+ this.getName());
+ }
Thread.currentThread().interrupt();
break;
}
@@ -86,7 +89,14 @@ public abstract class AbstractDriverThread extends Thread
implements Closeable {
try (SetThreadName driverTaskName =
new
SetThreadName(next.getDriver().getDriverTaskId().getFullId())) {
Throwable rootCause = ErrorHandlingCommonUtils.getRootCause(e);
- if (rootCause instanceof CorruptedTsFileException) {
+ if (closed && rootCause instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ // Still abort the task below so that shutdown releases its
query resources.
+ next.setAbortCause(
+ new DriverTaskAbortedException(
+ next.getDriverTaskId().getFullId(),
+ DriverTaskAbortedException.BY_SCHEDULER_SHUTDOWN));
+ } else if (rootCause instanceof CorruptedTsFileException) {
// CorruptedTsFileException no longer chains the original
IOException as its
// cause (it uses addSuppressed instead), so getRootCause
returns the exception
// itself and we can match it here.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskAbortedException.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskAbortedException.java
index fb683db9962..0f800376d33 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskAbortedException.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskAbortedException.java
@@ -34,6 +34,8 @@ public class DriverTaskAbortedException extends Exception {
DataNodeQueryMessages.DRIVER_TASK_ABORTED_BY_ALREADY_BEING_CANCELLED;
public static final String BY_INTERNAL_ERROR_SCHEDULED =
DataNodeQueryMessages.DRIVER_TASK_ABORTED_BY_INTERNAL_ERROR_SCHEDULED;
+ public static final String BY_SCHEDULER_SHUTDOWN =
+ DataNodeQueryMessages.EXCEPTION_SCHEDULER_SHUTDOWN_323938F5;
public static final String BY_MEMORY_NOT_ENOUGH =
DataNodeQueryMessages.DRIVER_TASK_ABORTED_BY_MEMORY_NOT_ENOUGH;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/ExecutionResult.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/ExecutionResult.java
index c00df6d8c73..2f46460635a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/ExecutionResult.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/ExecutionResult.java
@@ -26,8 +26,16 @@ public class ExecutionResult {
public final QueryId queryId;
public final TSStatus status;
+ /** Whether this failure originated from an interrupted wait rather than
another query error. */
+ public final boolean failureCausedByInterruption;
+
public ExecutionResult(QueryId queryId, TSStatus status) {
+ this(queryId, status, false);
+ }
+
+ ExecutionResult(QueryId queryId, TSStatus status, boolean
failureCausedByInterruption) {
this.queryId = queryId;
this.status = status;
+ this.failureCausedByInterruption = failureCausedByInterruption;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
index 5a094814607..b8f753d70a0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
@@ -644,19 +644,29 @@ public class QueryExecution implements IQueryExecution {
} catch (InterruptedException e) {
// TODO: (xingtanzjr) use more accurate error handling
Thread.currentThread().interrupt();
+ // Dispatch may already have converted an interruption to a failure
status. Preserve its
+ // recorded origin without reclassifying an unrelated failure.
+ TSStatus failureStatus = stateMachine.getFailureStatus();
+ if (failureStatus != null) {
+ return new ExecutionResult(
+ context.getQueryId(),
+ failureStatus,
+ stateMachine.isFailureCausedByInterruption(failureStatus));
+ }
return new ExecutionResult(
context.getQueryId(),
- stateMachine.getFailureStatus() == null
- ? RpcUtils.getStatus(
- TSStatusCode.INTERNAL_SERVER_ERROR,
stateMachine.getFailureMessage())
- : stateMachine.getFailureStatus());
+ RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR,
stateMachine.getFailureMessage()),
+ // A terminal query may have failed with an exception that has no
dedicated status code.
+ !stateMachine.getState().isDone());
} catch (ExecutionException e) {
+ TSStatus failureStatus = stateMachine.getFailureStatus();
return new ExecutionResult(
context.getQueryId(),
- stateMachine.getFailureStatus() == null
+ failureStatus == null
? RpcUtils.getStatus(
TSStatusCode.INTERNAL_SERVER_ERROR,
stateMachine.getFailureMessage())
- : stateMachine.getFailureStatus());
+ : failureStatus,
+ stateMachine.isFailureCausedByInterruption(failureStatus));
}
}
@@ -727,7 +737,8 @@ public class QueryExecution implements IQueryExecution {
||
IoTDBConstant.ClientVersion.V_1_0.equals(context.getSession().getVersion())) {
planner.setRedirectInfo(analysis, CONFIG.getAddressAndPort(), tsstatus);
}
- return new ExecutionResult(context.getQueryId(), tsstatus);
+ return new ExecutionResult(
+ context.getQueryId(), tsstatus,
stateMachine.isFailureCausedByInterruption(tsstatus));
}
public DistributedQueryPlan getDistributedPlan() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/ClusterScheduler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/ClusterScheduler.java
index 5e0aa02f5c0..4c828c3808c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/ClusterScheduler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/ClusterScheduler.java
@@ -125,7 +125,8 @@ public class ClusterScheduler implements IScheduler {
if (needRetry(result.getFailureStatus())) {
stateMachine.transitionToPendingRetry(result.getFailureStatus());
} else {
- stateMachine.transitionToFailed(result.getFailureStatus());
+ stateMachine.transitionToFailed(
+ result.getFailureStatus(),
result.isFailureCausedByInterruption());
}
return;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragInstanceDispatchResult.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragInstanceDispatchResult.java
index 6a621139455..dd9d39694e8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragInstanceDispatchResult.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragInstanceDispatchResult.java
@@ -26,13 +26,21 @@ public class FragInstanceDispatchResult {
private TSStatus failureStatus;
+ private final boolean failureCausedByInterruption;
+
public FragInstanceDispatchResult(boolean successful) {
this.successful = successful;
+ this.failureCausedByInterruption = false;
}
public FragInstanceDispatchResult(TSStatus failureStatus) {
+ this(failureStatus, false);
+ }
+
+ public FragInstanceDispatchResult(TSStatus failureStatus, boolean
failureCausedByInterruption) {
this.successful = false;
this.failureStatus = failureStatus;
+ this.failureCausedByInterruption = failureCausedByInterruption;
}
public boolean isSuccessful() {
@@ -42,4 +50,8 @@ public class FragInstanceDispatchResult {
public TSStatus getFailureStatus() {
return failureStatus;
}
+
+ public boolean isFailureCausedByInterruption() {
+ return failureCausedByInterruption;
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
index 24c42a0c16f..4894a8bc106 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java
@@ -330,7 +330,11 @@ public class FragmentInstanceDispatcherImpl implements
IFragInstanceDispatcher {
return immediateFuture(
new FragInstanceDispatchResult(
RpcUtils.getStatus(
- TSStatusCode.INTERNAL_SERVER_ERROR, "Interrupted errors: " +
e.getMessage())));
+ TSStatusCode.INTERNAL_SERVER_ERROR,
+ String.format(
+
DataNodeQueryMessages.MESSAGE_INTERRUPTED_ERRORS_ARG_876F1B4B,
+ e.getMessage())),
+ true));
}
if (dispatchFailures.isEmpty()) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
index 1630175738f..ad13233dbac 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
@@ -25,10 +25,12 @@ import org.apache.iotdb.commons.audit.UserEntity;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.ConfigRegionId;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.queryengine.common.SessionInfo;
import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.confignode.rpc.thrift.TGetDatabaseReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
@@ -72,6 +74,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -81,24 +84,27 @@ public class IoTDBInternalLocalReporter extends
IoTDBInternalReporter {
private static final Logger LOGGER =
LoggerFactory.getLogger(IoTDBInternalLocalReporter.class);
private static final SessionManager SESSION_MANAGER =
SessionManager.getInstance();
- private static final Coordinator COORDINATOR = Coordinator.getInstance();
+ private final Coordinator coordinator;
private final SessionInfo sessionInfo;
private final IPartitionFetcher partitionFetcher;
private final ISchemaFetcher schemaFetcher;
private Future<?> currentServiceFuture;
- private final ScheduledExecutorService service =
Executors.newSingleThreadScheduledExecutor();
+ private final ScheduledExecutorService service;
+ private volatile boolean stopped;
public IoTDBInternalLocalReporter() {
- partitionFetcher = ClusterPartitionFetcher.getInstance();
- schemaFetcher = ClusterSchemaFetcher.getInstance();
- sessionInfo =
+ this(
+ Executors.newSingleThreadScheduledExecutor(),
+ Coordinator.getInstance(),
new SessionInfo(
0,
new UserEntity(
AuthorityChecker.SUPER_USER_ID,
AuthorityChecker.SUPER_USER,
IoTDBDescriptor.getInstance().getConfig().getInternalAddress()),
- ZoneId.systemDefault());
+ ZoneId.systemDefault()),
+ ClusterPartitionFetcher.getInstance(),
+ ClusterSchemaFetcher.getInstance());
IClientManager<ConfigRegionId, ConfigNodeClient> configNodeClientManager =
ConfigNodeClientManager.getInstance();
@@ -125,13 +131,29 @@ public class IoTDBInternalLocalReporter extends
IoTDBInternalReporter {
}
}
+ IoTDBInternalLocalReporter(
+ ScheduledExecutorService service,
+ Coordinator coordinator,
+ SessionInfo sessionInfo,
+ IPartitionFetcher partitionFetcher,
+ ISchemaFetcher schemaFetcher) {
+ this.service = service;
+ this.coordinator = coordinator;
+ this.sessionInfo = sessionInfo;
+ this.partitionFetcher = partitionFetcher;
+ this.schemaFetcher = schemaFetcher;
+ }
+
@Override
public InternalReporterType getType() {
return InternalReporterType.IOTDB;
}
@Override
- public boolean start() {
+ public synchronized boolean start() {
+ if (isStopping()) {
+ return false;
+ }
if (currentServiceFuture != null) {
LOGGER.warn(DataNodeMiscMessages.INTERNAL_REPORTER_ALREADY_STARTED);
return false;
@@ -148,7 +170,9 @@ public class IoTDBInternalLocalReporter extends
IoTDBInternalReporter {
}
@Override
- public boolean stop() {
+ public synchronized boolean stop() {
+ // Publish this before interrupting an in-flight write, including during
reporter reload.
+ stopped = true;
if (currentServiceFuture != null) {
currentServiceFuture.cancel(true);
currentServiceFuture = null;
@@ -166,32 +190,86 @@ public class IoTDBInternalLocalReporter extends
IoTDBInternalReporter {
@Override
protected void writeMetricToIoTDB(Map<String, Object> valueMap, String
prefix, long time) {
- service.execute(
- () -> {
- try {
- TSStatus result = insertRecord(valueMap, prefix, time);
- if (result.getCode() ==
TSStatusCode.METADATA_ERROR.getStatusCode()) {
- createTimeSeries(valueMap, prefix);
- result = insertRecord(valueMap, prefix, time);
- }
- if (result.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn(DataNodeMiscMessages.FAILED_UPDATE_METRIC_VALUE,
result);
- }
- } catch (IoTDBConnectionException e1) {
- LOGGER.warn(
- DataNodeMiscMessages
-
.MISC_LOG_FAILED_TO_UPDATE_THE_VALUE_OF_METRIC_BECAUSE_OF_CONNECTION_B0FC4929,
- e1);
- } catch (IllegalPathException | QueryProcessException e2) {
- LOGGER.warn(
- DataNodeMiscMessages
-
.MISC_LOG_FAILED_TO_UPDATE_THE_VALUE_OF_METRIC_BECAUSE_OF_INTERNAL_E5C64806,
- e2);
- }
- });
+ if (isStopping()) {
+ return;
+ }
+ try {
+ service.execute(() -> writeMetric(valueMap, prefix, time));
+ } catch (RejectedExecutionException e) {
+ // stop() can shut down the executor between the check above and
submission.
+ if (!stopped) {
+ throw e;
+ }
+ }
}
- private TSStatus insertRecord(Map<String, Object> valueMap, String prefix,
long time)
+ private void writeMetric(Map<String, Object> valueMap, String prefix, long
time) {
+ if (isStopping()) {
+ return;
+ }
+ try {
+ ExecutionResult result = insertRecord(valueMap, prefix, time);
+ if (result.status.getCode() ==
TSStatusCode.METADATA_ERROR.getStatusCode() && !isStopping()) {
+ createTimeSeries(valueMap, prefix);
+ if (isStopping()) {
+ return;
+ }
+ result = insertRecord(valueMap, prefix, time);
+ }
+ if (result.status.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ if (isExpectedShutdownFailure(result)) {
+ LOGGER.debug(DataNodeMiscMessages.FAILED_UPDATE_METRIC_VALUE,
result.status);
+ } else {
+ LOGGER.warn(DataNodeMiscMessages.FAILED_UPDATE_METRIC_VALUE,
result.status);
+ }
+ }
+ } catch (IoTDBConnectionException e1) {
+ if (isInterruptedDuringShutdown(e1)) {
+ LOGGER.debug(
+ DataNodeMiscMessages
+
.MISC_LOG_FAILED_TO_UPDATE_THE_VALUE_OF_METRIC_BECAUSE_OF_CONNECTION_B0FC4929,
+ e1);
+ } else {
+ LOGGER.warn(
+ DataNodeMiscMessages
+
.MISC_LOG_FAILED_TO_UPDATE_THE_VALUE_OF_METRIC_BECAUSE_OF_CONNECTION_B0FC4929,
+ e1);
+ }
+ } catch (IllegalPathException | QueryProcessException e2) {
+ if (isInterruptedDuringShutdown(e2)) {
+ LOGGER.debug(
+ DataNodeMiscMessages
+
.MISC_LOG_FAILED_TO_UPDATE_THE_VALUE_OF_METRIC_BECAUSE_OF_INTERNAL_E5C64806,
+ e2);
+ } else {
+ LOGGER.warn(
+ DataNodeMiscMessages
+
.MISC_LOG_FAILED_TO_UPDATE_THE_VALUE_OF_METRIC_BECAUSE_OF_INTERNAL_E5C64806,
+ e2);
+ }
+ }
+ }
+
+ private boolean isStopping() {
+ return stopped || CommonDescriptor.getInstance().getConfig().isStopping();
+ }
+
+ private boolean isExpectedShutdownFailure(ExecutionResult result) {
+ return isStopping()
+ && (result.failureCausedByInterruption
+ || (CommonDescriptor.getInstance().getConfig().isStopping()
+ && result.status.getCode() ==
TSStatusCode.SYSTEM_READ_ONLY.getStatusCode()));
+ }
+
+ private boolean isInterruptedDuringShutdown(Exception e) {
+ if (isStopping() && ErrorHandlingCommonUtils.getRootCause(e) instanceof
InterruptedException) {
+ Thread.currentThread().interrupt();
+ return true;
+ }
+ return false;
+ }
+
+ private ExecutionResult insertRecord(Map<String, Object> valueMap, String
prefix, long time)
throws IoTDBConnectionException, QueryProcessException,
IllegalPathException {
TSInsertRecordReq request = new TSInsertRecordReq();
List<String> measurements = new ArrayList<>();
@@ -214,10 +292,8 @@ public class IoTDBInternalLocalReporter extends
IoTDBInternalReporter {
InsertRowStatement s = StatementGenerator.createStatement(request);
final long queryId = SESSION_MANAGER.requestQueryId();
- ExecutionResult result =
- COORDINATOR.executeForTreeModel(
- s, queryId, sessionInfo, "", partitionFetcher, schemaFetcher);
- return result.status;
+ return coordinator.executeForTreeModel(
+ s, queryId, sessionInfo, "", partitionFetcher, schemaFetcher);
}
private void createTimeSeries(Map<String, Object> valueMap, String prefix)
@@ -244,10 +320,14 @@ public class IoTDBInternalLocalReporter extends
IoTDBInternalReporter {
final long queryId = SESSION_MANAGER.requestQueryId();
ExecutionResult result =
- COORDINATOR.executeForTreeModel(
+ coordinator.executeForTreeModel(
s, queryId, sessionInfo, "", partitionFetcher, schemaFetcher);
if (result.status.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn(DataNodeMiscMessages.FAILED_AUTO_CREATE_TIMESERIES, paths,
result.status);
+ if (isExpectedShutdownFailure(result)) {
+ LOGGER.debug(DataNodeMiscMessages.FAILED_AUTO_CREATE_TIMESERIES,
paths, result.status);
+ } else {
+ LOGGER.warn(DataNodeMiscMessages.FAILED_AUTO_CREATE_TIMESERIES, paths,
result.status);
+ }
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachineTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachineTest.java
index ad7ab094748..118ff7b5025 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachineTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/QueryStateMachineTest.java
@@ -31,13 +31,17 @@ import org.junit.Assert;
import org.junit.Test;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
public class QueryStateMachineTest {
@@ -158,6 +162,37 @@ public class QueryStateMachineTest {
stateMachine.getFailureStatus().code);
}
+ @Test
+ public void testInterruptionProvenanceBelongsToTheRecordedStatus() {
+ QueryStateMachine stateMachine =
+ new QueryStateMachine(genQueryId(), mock(ExecutorService.class));
+ TSStatus interruption = new
TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+
+ stateMachine.transitionToFailed(interruption, true);
+
+ assertSame(interruption, stateMachine.getFailureStatus());
+ assertTrue(stateMachine.isFailureCausedByInterruption(interruption));
+ assertFalse(
+ stateMachine.isFailureCausedByInterruption(
+ new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode())));
+ }
+
+ @Test
+ public void testLaterInterruptionDoesNotReclassifyExistingFailure() {
+ QueryStateMachine stateMachine =
+ new QueryStateMachine(genQueryId(), mock(ExecutorService.class));
+ TSStatus original = new
TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode());
+ TSStatus interruption = new
TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+
+ stateMachine.transitionToFailed(original);
+ stateMachine.transitionToFailed(interruption, true);
+ stateMachine.transitionToFailed(original, true);
+
+ assertSame(original, stateMachine.getFailureStatus());
+ assertFalse(stateMachine.isFailureCausedByInterruption(original));
+ assertFalse(stateMachine.isFailureCausedByInterruption(interruption));
+ }
+
private QueryStateMachine genQueryStateMachine() {
return new QueryStateMachine(
genQueryId(),
IoTDBThreadPoolFactory.newSingleThreadExecutor("TestQueryStateMachine"));
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThreadShutdownTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThreadShutdownTest.java
new file mode 100644
index 00000000000..8f57718c665
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThreadShutdownTest.java
@@ -0,0 +1,271 @@
+/*
+ * 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.schedule;
+
+import org.apache.iotdb.calc.execution.schedule.queue.IndexedBlockingQueue;
+import org.apache.iotdb.commons.exception.QueryTimeoutException;
+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.IDriver;
+import
org.apache.iotdb.db.queryengine.execution.schedule.queue.L1PriorityQueue;
+import org.apache.iotdb.db.queryengine.execution.schedule.task.DriverTask;
+import org.apache.iotdb.db.queryengine.execution.schedule.task.DriverTaskId;
+import
org.apache.iotdb.db.queryengine.execution.schedule.task.DriverTaskStatus;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class AbstractDriverThreadShutdownTest {
+
+ private final Logger logger = (Logger)
LoggerFactory.getLogger(AbstractDriverThread.class);
+ private final ListAppender<ILoggingEvent> appender = new ListAppender<>();
+ private final ITaskScheduler scheduler = mock(ITaskScheduler.class);
+ private final ThreadProducer producer = mock(ThreadProducer.class);
+ private final ThreadGroup threadGroup = new ThreadGroup("shutdown-test");
+ private Level originalLevel;
+
+ @Before
+ public void setUp() {
+ originalLevel = logger.getLevel();
+ logger.setLevel(Level.DEBUG);
+ appender.start();
+ logger.addAppender(appender);
+ }
+
+ @After
+ public void tearDown() {
+ logger.detachAppender(appender);
+ appender.stop();
+ logger.setLevel(originalLevel);
+ }
+
+ @Test
+ public void testCloseWhilePolling() throws Exception {
+ CountDownLatch polling = new CountDownLatch(1);
+ IndexedBlockingQueue<DriverTask> queue = pollingQueue(polling);
+ AbstractDriverThread thread =
+ new DriverTaskTimeoutSentinelThread("sentinel", threadGroup, queue,
scheduler, producer);
+
+ interruptAfterEntered(thread, polling, true);
+
+ assertFalse(hasWarning());
+ verify(scheduler, never()).toAborted(any());
+ verify(producer, never()).produce(anyString(), any(), any(), any());
+ }
+
+ @Test
+ public void testUnexpectedPollingInterruptStillWarnsAndReplacesThread()
throws Exception {
+ CountDownLatch polling = new CountDownLatch(1);
+ IndexedBlockingQueue<DriverTask> queue = pollingQueue(polling);
+ AbstractDriverThread thread =
+ new DriverTaskTimeoutSentinelThread("sentinel", threadGroup, queue,
scheduler, producer);
+
+ interruptAfterEntered(thread, polling, false);
+
+ assertTrue(hasWarning());
+ verify(producer).produce("sentinel", threadGroup, queue, producer);
+ }
+
+ @Test
+ public void testCloseWhileSentinelWaitsStillAbortsTask() throws Exception {
+ IndexedBlockingQueue<DriverTask> queue = pollingQueue(new
CountDownLatch(0));
+ DriverTask task = newTask(60_000);
+ queue.push(task);
+ CountDownLatch executing = new CountDownLatch(1);
+ AbstractDriverThread thread = sentinel(queue, executing);
+ AtomicBoolean interruptedDuringAbort = new AtomicBoolean();
+ doAnswer(
+ invocation -> {
+
interruptedDuringAbort.set(Thread.currentThread().isInterrupted());
+ return null;
+ })
+ .when(scheduler)
+ .toAborted(task);
+
+ interruptAfterEntered(thread, executing, true);
+
+ assertFalse(hasWarning());
+ assertTrue(interruptedDuringAbort.get());
+ assertAbortCause(task, DriverTaskAbortedException.BY_SCHEDULER_SHUTDOWN);
+ verify(scheduler).toAborted(task);
+ verify(scheduler, never()).enforceTimeLimit(any());
+ verify(producer, never()).produce(anyString(), any(), any(), any());
+ }
+
+ @Test
+ public void testUnexpectedSentinelInterruptStillAbortsAndWarns() throws
Exception {
+ IndexedBlockingQueue<DriverTask> queue = pollingQueue(new
CountDownLatch(0));
+ DriverTask task = newTask(60_000);
+ queue.push(task);
+ CountDownLatch executing = new CountDownLatch(1);
+ AbstractDriverThread thread = sentinel(queue, executing);
+ doAnswer(
+ invocation -> {
+ // End the test only after the running thread has handled the
unexpected interrupt.
+ thread.close();
+ return null;
+ })
+ .when(scheduler)
+ .toAborted(task);
+
+ interruptAfterEntered(thread, executing, false);
+
+ assertTrue(hasWarning());
+ assertAbortCause(task,
DriverTaskAbortedException.BY_INTERNAL_ERROR_SCHEDULED);
+ verify(scheduler).toAborted(task);
+ }
+
+ @Test
+ public void testWrappedShutdownInterruptStillAbortsTask() throws Exception {
+ verifyFailureDuringShutdown(new RuntimeException(new
InterruptedException()), false);
+ }
+
+ @Test
+ public void testNonInterruptFailureDuringShutdownStillWarns() throws
Exception {
+ verifyFailureDuringShutdown(new IllegalStateException("mock failure"),
true);
+ }
+
+ @Test
+ public void testSentinelStillEnforcesTimeout() throws Exception {
+ DriverTask task = newTask(0);
+ DriverTaskTimeoutSentinelThread thread =
+ new DriverTaskTimeoutSentinelThread(
+ "sentinel", threadGroup, pollingQueue(new CountDownLatch(0)),
scheduler, producer);
+
+ thread.execute(task);
+
+ assertTrue(task.getAbortCause().get() instanceof QueryTimeoutException);
+ verify(scheduler).toAborted(task);
+ verify(scheduler, never()).enforceTimeLimit(any());
+ }
+
+ private void verifyFailureDuringShutdown(RuntimeException failure, boolean
shouldWarn)
+ throws Exception {
+ IndexedBlockingQueue<DriverTask> queue = pollingQueue(new
CountDownLatch(0));
+ DriverTask task = newTask(60_000);
+ queue.push(task);
+ CountDownLatch executing = new CountDownLatch(1);
+ AbstractDriverThread thread =
+ new AbstractDriverThread("worker", threadGroup, queue, scheduler,
producer) {
+ @Override
+ protected void execute(DriverTask task) {
+ executing.countDown();
+ try {
+ new CountDownLatch(1).await();
+ } catch (InterruptedException e) {
+ throw failure;
+ }
+ }
+ };
+
+ interruptAfterEntered(thread, executing, true);
+
+ assertEquals(shouldWarn, hasWarning());
+ assertAbortCause(
+ task,
+ shouldWarn
+ ? DriverTaskAbortedException.BY_INTERNAL_ERROR_SCHEDULED
+ : DriverTaskAbortedException.BY_SCHEDULER_SHUTDOWN);
+ verify(scheduler).toAborted(task);
+ }
+
+ private AbstractDriverThread sentinel(
+ IndexedBlockingQueue<DriverTask> queue, CountDownLatch executing) {
+ return new DriverTaskTimeoutSentinelThread(
+ "sentinel", threadGroup, queue, scheduler, producer) {
+ @Override
+ public void execute(DriverTask task) throws InterruptedException {
+ executing.countDown();
+ super.execute(task);
+ }
+ };
+ }
+
+ private IndexedBlockingQueue<DriverTask> pollingQueue(CountDownLatch
polling) {
+ return new L1PriorityQueue<DriverTask>(
+ 10, new DriverTask.TimeoutComparator(), new DriverTask()) {
+ @Override
+ public DriverTask poll() throws InterruptedException {
+ polling.countDown();
+ return super.poll();
+ }
+ };
+ }
+
+ private DriverTask newTask(long timeoutMs) {
+ IDriver driver = mock(IDriver.class);
+ when(driver.getDriverTaskId())
+ .thenReturn(
+ new DriverTaskId(
+ new FragmentInstanceId(new PlanFragmentId(new QueryId("test"),
0), "instance"), 0));
+ return new DriverTask(driver, timeoutMs, DriverTaskStatus.READY, null, 0,
false);
+ }
+
+ private void interruptAfterEntered(
+ AbstractDriverThread thread, CountDownLatch entered, boolean close)
throws Exception {
+ thread.setDaemon(true);
+ thread.start();
+ try {
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
+ if (close) {
+ thread.close();
+ }
+ thread.interrupt();
+ thread.join(10_000);
+ assertFalse("Driver thread did not exit", thread.isAlive());
+ } finally {
+ thread.close();
+ thread.interrupt();
+ thread.join(10_000);
+ }
+ }
+
+ private void assertAbortCause(DriverTask task, String reason) {
+ assertEquals(
+ new DriverTaskAbortedException(task.getDriverTaskId().getFullId(),
reason).getMessage(),
+ task.getAbortCause().get().getMessage());
+ }
+
+ private boolean hasWarning() {
+ return appender.list.stream().anyMatch(event ->
event.getLevel().isGreaterOrEqual(Level.WARN));
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
index 959bd4b9d6a..0c154850ca5 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern;
+import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseManager;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import com.google.common.collect.ImmutableList;
@@ -65,6 +66,9 @@ public class TreeViewTest {
@Before
public void setup() {
+ // This planner test has no ConfigNode heartbeats and may exceed the
default lease timeout.
+ MetadataLeaseManager.getInstance().updateFenceThresholdMs(Long.MAX_VALUE);
+ MetadataLeaseManager.getInstance().recoveryLeaseForTest(true);
TsTable tsTable = new TsTable(DEVICE_VIEW_TEST_TABLE);
tsTable.addProp(TsTable.TTL_PROPERTY, Long.MAX_VALUE + "");
tsTable.addProp(TreeViewSchema.TREE_PATH_PATTERN, "root.test" + ".**");
@@ -74,7 +78,13 @@ public class TreeViewTest {
@After
public void tearDown() {
- DataNodeTableCache.getInstance().invalid(TREE_VIEW_DB);
+ try {
+ DataNodeTableCache.getInstance().invalid(TREE_VIEW_DB);
+ } finally {
+ // Renew before restoring the timeout to avoid racing the background
lease check.
+ MetadataLeaseManager.getInstance().recoveryLeaseForTest(true);
+ MetadataLeaseManager.getInstance().updateFenceThresholdMs(20_000);
+ }
}
@Test
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporterTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporterTest.java
new file mode 100644
index 00000000000..a94cefc54c0
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporterTest.java
@@ -0,0 +1,623 @@
+/*
+ * 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.service.metrics;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.IClientManager;
+import
org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.queryengine.common.SessionInfo;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.execution.QueryStateMachine;
+import org.apache.iotdb.db.queryengine.plan.Coordinator;
+import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis;
+import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
+import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
+import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
+import org.apache.iotdb.db.queryengine.plan.execution.QueryExecution;
+import org.apache.iotdb.db.queryengine.plan.planner.IPlanner;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.PlanFragment;
+import org.apache.iotdb.db.queryengine.plan.scheduler.ClusterScheduler;
+import org.apache.iotdb.db.queryengine.plan.statement.Statement;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement;
+import
org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateMultiTimeSeriesStatement;
+import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.stubbing.OngoingStubbing;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+public class IoTDBInternalLocalReporterTest {
+
+ private static int originalDataNodeId;
+
+ private final Logger logger = (Logger)
LoggerFactory.getLogger(IoTDBInternalLocalReporter.class);
+ private final ListAppender<ILoggingEvent> appender = new ListAppender<>();
+ private final Coordinator coordinator = mock(Coordinator.class);
+ private final SessionInfo sessionInfo = mock(SessionInfo.class);
+ private final IPartitionFetcher partitionFetcher =
mock(IPartitionFetcher.class);
+ private final ISchemaFetcher schemaFetcher = mock(ISchemaFetcher.class);
+ private ScheduledExecutorService service;
+ private IoTDBInternalLocalReporter reporter;
+ private Level originalLevel;
+ private boolean originalStopping;
+
+ @BeforeClass
+ public static void initializeDataNodeId() {
+ originalDataNodeId =
IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+ IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0);
+ }
+
+ @AfterClass
+ public static void restoreDataNodeId() {
+
IoTDBDescriptor.getInstance().getConfig().setDataNodeId(originalDataNodeId);
+ }
+
+ @Before
+ public void setUp() {
+ originalLevel = logger.getLevel();
+ logger.setLevel(Level.DEBUG);
+ appender.start();
+ logger.addAppender(appender);
+ originalStopping = CommonDescriptor.getInstance().getConfig().isStopping();
+ CommonDescriptor.getInstance().getConfig().setStopping(false);
+ service = mock(ScheduledExecutorService.class);
+ doAnswer(
+ invocation -> {
+ invocation.<Runnable>getArgument(0).run();
+ return null;
+ })
+ .when(service)
+ .execute(any(Runnable.class));
+ reporter = newReporter(service);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ // Some tests intentionally restore the interrupted flag while returning a
failed status.
+ Thread.interrupted();
+ reporter.stop();
+ service.awaitTermination(5, TimeUnit.SECONDS);
+ CommonDescriptor.getInstance().getConfig().setStopping(originalStopping);
+ logger.detachAppender(appender);
+ appender.stop();
+ logger.setLevel(originalLevel);
+ }
+
+ @Test
+ public void testStoppedReporterRejectsNewWritesAndRestart() {
+ reporter.stop();
+
+ writeMetric();
+
+ assertFalse(reporter.start());
+ verify(service, never()).execute(any(Runnable.class));
+ verifyZeroInteractions(coordinator);
+ assertFalse(hasWarning());
+ }
+
+ @Test
+ public void testNodeShutdownRejectsNewWrites() {
+ CommonDescriptor.getInstance().getConfig().setStopping(true);
+
+ writeMetric();
+
+ verify(service, never()).execute(any(Runnable.class));
+ verifyZeroInteractions(coordinator);
+ assertFalse(hasWarning());
+ }
+
+ @Test
+ public void testQueuedWriteDoesNotRunAfterStop() {
+ doNothing().when(service).execute(any(Runnable.class));
+ writeMetric();
+ ArgumentCaptor<Runnable> task = ArgumentCaptor.forClass(Runnable.class);
+ verify(service).execute(task.capture());
+
+ reporter.stop();
+ task.getValue().run();
+
+ verifyZeroInteractions(coordinator);
+ assertFalse(hasWarning());
+ }
+
+ @Test
+ public void testStopBetweenCheckAndSubmission() {
+ doAnswer(
+ invocation -> {
+ reporter.stop();
+ throw new RejectedExecutionException("executor stopped");
+ })
+ .when(service)
+ .execute(any(Runnable.class));
+
+ writeMetric();
+
+ verifyZeroInteractions(coordinator);
+ assertFalse(hasWarning());
+ }
+
+ @Test(expected = RejectedExecutionException.class)
+ public void testUnexpectedSubmissionFailureIsPropagated() {
+ doThrow(new RejectedExecutionException("unexpected rejection"))
+ .when(service)
+ .execute(any(Runnable.class));
+
+ writeMetric();
+ }
+
+ @Test
+ public void testMissingTimeseriesIsCreatedAndWriteRetried() {
+ execution()
+ .thenReturn(
+ result(TSStatusCode.METADATA_ERROR),
+ result(TSStatusCode.SUCCESS_STATUS),
+ result(TSStatusCode.SUCCESS_STATUS));
+
+ writeMetric();
+
+ List<Statement> statements = executedStatements(3);
+ assertTrue(statements.get(0) instanceof InsertRowStatement);
+ assertTrue(statements.get(1) instanceof CreateMultiTimeSeriesStatement);
+ assertTrue(statements.get(2) instanceof InsertRowStatement);
+ assertFalse(hasWarning());
+ }
+
+ @Test
+ public void testStopAfterInsertDoesNotCreateTimeseries() {
+ execution()
+ .thenAnswer(
+ invocation -> {
+ reporter.stop();
+ Thread.currentThread().interrupt();
+ return result(TSStatusCode.METADATA_ERROR);
+ });
+
+ writeMetric();
+
+ executedStatements(1);
+ assertTrue(Thread.currentThread().isInterrupted());
+ assertTrue(hasWarning());
+ }
+
+ @Test
+ public void testStopInterruptsAutoCreateAndPreventsRetry() throws Exception {
+ verifyReloadInterrupt(null, null, true);
+ }
+
+ @Test
+ public void testNodeShutdownReadOnlyFailureDoesNotWarn() {
+ execution()
+ .thenAnswer(
+ invocation -> {
+ CommonDescriptor.getInstance().getConfig().setStopping(true);
+ return result(TSStatusCode.SYSTEM_READ_ONLY);
+ });
+
+ writeMetric();
+
+ executedStatements(1);
+ assertFalse(hasWarning());
+ }
+
+ @Test
+ public void testReadOnlyFailureWithoutNodeShutdownStillWarns() {
+ execution().thenReturn(result(TSStatusCode.SYSTEM_READ_ONLY));
+
+ writeMetric();
+
+ assertTrue(hasWarning());
+ }
+
+ @Test
+ public void testInterruptedWaitWithoutShutdownStillWarns() throws Exception {
+ QueryExecution queryExecution =
+ queryExecutionWithDelayedStatus(null, null, new CountDownLatch(0));
+ execution()
+ .thenAnswer(
+ invocation -> {
+ Thread.currentThread().interrupt();
+ return queryExecution.getStatus();
+ });
+
+ writeMetric();
+
+ assertTrue(hasWarning());
+ }
+
+ @Test
+ public void testInternalErrorDuringStopWithoutInterruptionStillWarns() {
+ execution()
+ .thenAnswer(
+ invocation -> {
+ reporter.stop();
+ return result(TSStatusCode.INTERNAL_SERVER_ERROR);
+ });
+
+ writeMetric();
+
+ assertTrue(hasWarning());
+ }
+
+ @Test
+ public void testAutoCreateFailureWithoutInterruptionStillWarnsDuringStop() {
+ execution()
+ .thenReturn(result(TSStatusCode.METADATA_ERROR))
+ .thenAnswer(
+ invocation -> {
+ reporter.stop();
+ return result(TSStatusCode.INTERNAL_SERVER_ERROR);
+ });
+
+ writeMetric();
+
+ executedStatements(2);
+ assertTrue(hasWarning());
+ }
+
+ @Test
+ public void testReporterReloadDoesNotHideReadOnlyFailure() {
+ execution()
+ .thenAnswer(
+ invocation -> {
+ reporter.stop();
+ return result(TSStatusCode.SYSTEM_READ_ONLY);
+ });
+
+ writeMetric();
+
+ assertTrue(hasWarning());
+ }
+
+ @Test
+ public void testReloadInterruptPreservesRecordedReadOnlyFailure() throws
Exception {
+ verifyReloadInterrupt(new
TSStatus(TSStatusCode.SYSTEM_READ_ONLY.getStatusCode()), null, false);
+ }
+
+ @Test
+ public void testReloadInterruptPreservesRecordedWriteFailure() throws
Exception {
+ verifyReloadInterrupt(
+ new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode()), null,
false);
+ }
+
+ @Test
+ public void testReloadInterruptPreservesRecordedInternalFailure() throws
Exception {
+ verifyReloadInterrupt(
+ new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()),
null, false);
+ }
+
+ @Test
+ public void testReloadInterruptPreservesRecordedException() throws Exception
{
+ verifyReloadInterrupt(null, new IllegalStateException("existing write
failure"), false);
+ }
+
+ @Test
+ public void testReloadInterruptPreservesRecordedAutoCreateFailure() throws
Exception {
+ verifyReloadInterrupt(
+ new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode()), null,
true);
+ }
+
+ @Test
+ public void testReloadWithOnlyInterruptedWaitDoesNotWarn() throws Exception {
+ verifyReloadInterrupt(null, null, false);
+ }
+
+ @Test
+ public void testRemoteWriteInterruptedDuringReloadDoesNotWarn() throws
Exception {
+ verifyRemoteDispatchInterruption(false, false, null);
+ }
+
+ @Test
+ public void testRemoteAutoCreateInterruptedDuringReloadDoesNotWarn() throws
Exception {
+ verifyRemoteDispatchInterruption(true, false, null);
+ }
+
+ @Test
+ public void testRemoteInterruptionSurvivesClearedThreadFlag() throws
Exception {
+ verifyRemoteDispatchInterruption(false, true, null);
+ }
+
+ @Test
+ public void testRemoteInterruptionDoesNotReplaceExistingFailure() throws
Exception {
+ verifyRemoteDispatchInterruption(
+ false, false, new
TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode()));
+ }
+
+ @Test
+ public void testRemoteInterruptionDoesNotHideExistingException() throws
Exception {
+ verifyRemoteDispatchInterruption(
+ false, false, null, new IllegalStateException("existing write
failure"));
+ }
+
+ private void verifyRemoteDispatchInterruption(
+ boolean autoCreate, boolean clearInterrupt, TSStatus recordedFailure)
throws Exception {
+ verifyRemoteDispatchInterruption(autoCreate, clearInterrupt,
recordedFailure, null);
+ }
+
+ private void verifyRemoteDispatchInterruption(
+ boolean autoCreate,
+ boolean clearInterrupt,
+ TSStatus recordedFailure,
+ Throwable recordedException)
+ throws Exception {
+ service = Executors.newSingleThreadScheduledExecutor();
+ reporter = newReporter(service);
+ MPPQueryContext context = mock(MPPQueryContext.class);
+ when(context.getQueryId()).thenReturn(new QueryId("remote_metric"));
+ IPlanner planner = mock(IPlanner.class);
+ when(planner.analyze(context)).thenReturn(mock(IAnalysis.class));
+ ExecutorService notifications = mock(ExecutorService.class);
+ QueryExecution queryExecution = new QueryExecution(planner, context,
notifications);
+ QueryStateMachine stateMachine = stateMachine(queryExecution);
+ if (recordedFailure != null) {
+ stateMachine.transitionToFailed(recordedFailure);
+ } else if (recordedException != null) {
+ stateMachine.transitionToFailed(recordedException);
+ } else {
+ stateMachine.transitionToPlanned();
+ }
+
+ TEndPoint endpoint = new TEndPoint("192.0.2.1", 10730);
+ TDataNodeLocation remote = new
TDataNodeLocation().setInternalEndPoint(endpoint);
+ TRegionReplicaSet region =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 1),
+ Collections.singletonList(remote));
+ PlanNode node = mock(PlanNode.class);
+ when(node.serializeToByteBuffer()).thenReturn(ByteBuffer.allocate(0));
+ when(node.getChildren()).thenReturn(Collections.emptyList());
+ PlanFragment fragment = mock(PlanFragment.class);
+ when(fragment.getPlanNodeTree()).thenReturn(node);
+ FragmentInstance instance = mock(FragmentInstance.class);
+ when(instance.getHostDataNode()).thenReturn(remote);
+ when(instance.getRegionReplicaSet()).thenReturn(region);
+ when(instance.getFragment()).thenReturn(fragment);
+ DistributedQueryPlan plan = mock(DistributedQueryPlan.class);
+ when(plan.getInstances()).thenReturn(Collections.singletonList(instance));
+
+ @SuppressWarnings("unchecked")
+ IClientManager<TEndPoint, AsyncDataNodeInternalServiceClient> clients =
+ mock(IClientManager.class);
+ AsyncDataNodeInternalServiceClient client =
mock(AsyncDataNodeInternalServiceClient.class);
+ when(clients.borrowClient(endpoint)).thenReturn(client);
+ CountDownLatch dispatched = new CountDownLatch(1);
+ doAnswer(
+ invocation -> {
+ // Leave the RPC response pending so the real
AsyncPlanNodeSender waits.
+ dispatched.countDown();
+ return null;
+ })
+ .when(client)
+ .sendBatchPlanNode(any(TSendBatchPlanNodeReq.class), any());
+ ClusterScheduler scheduler =
+ new ClusterScheduler(context, stateMachine, plan, QueryType.WRITE,
null, null, clients);
+ AtomicReference<ExecutionResult> returnedResult = new AtomicReference<>();
+ OngoingStubbing<ExecutionResult> execution = execution();
+ if (autoCreate) {
+ execution = execution.thenReturn(result(TSStatusCode.METADATA_ERROR));
+ }
+ execution.thenAnswer(
+ invocation -> {
+ scheduler.start();
+ if (clearInterrupt) {
+ Thread.interrupted();
+ }
+ // Deliver the getStatus() listener immediately, also exercising its
normal return
+ // path when the interrupt flag has been consumed after dispatch.
+ doAnswer(
+ notification -> {
+ notification.<Runnable>getArgument(0).run();
+ return null;
+ })
+ .when(notifications)
+ .execute(any(Runnable.class));
+ ExecutionResult result = queryExecution.getStatus();
+ returnedResult.set(result);
+ return result;
+ });
+
+ writeMetric();
+ assertTrue(dispatched.await(5, TimeUnit.SECONDS));
+ reporter.stop();
+ assertTrue(service.awaitTermination(5, TimeUnit.SECONDS));
+
+ ExecutionResult result = returnedResult.get();
+ if (recordedFailure != null) {
+ assertSame(recordedFailure, result.status);
+ } else {
+ assertEquals(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode(),
result.status.getCode());
+ }
+ boolean hasRecordedFailure = recordedFailure != null || recordedException
!= null;
+ assertEquals(hasRecordedFailure, hasWarning());
+ assertEquals(!hasRecordedFailure, result.failureCausedByInterruption);
+ executedStatements(autoCreate ? 2 : 1);
+ verify(client).sendBatchPlanNode(any(TSendBatchPlanNodeReq.class), any());
+ }
+
+ private void verifyReloadInterrupt(
+ TSStatus recordedFailure, Throwable recordedException, boolean
autoCreate) throws Exception {
+ service = Executors.newSingleThreadScheduledExecutor();
+ reporter = newReporter(service);
+ CountDownLatch waitingForStatus = new CountDownLatch(1);
+ QueryExecution queryExecution =
+ queryExecutionWithDelayedStatus(recordedFailure, recordedException,
waitingForStatus);
+ AtomicReference<ExecutionResult> returnedResult = new AtomicReference<>();
+ AtomicBoolean interrupted = new AtomicBoolean();
+ OngoingStubbing<ExecutionResult> execution = execution();
+ if (autoCreate) {
+ execution = execution.thenReturn(result(TSStatusCode.METADATA_ERROR));
+ }
+ execution.thenAnswer(
+ invocation -> {
+ ExecutionResult result = queryExecution.getStatus();
+ returnedResult.set(result);
+ interrupted.set(Thread.currentThread().isInterrupted());
+ return result;
+ });
+
+ writeMetric();
+ assertTrue(waitingForStatus.await(5, TimeUnit.SECONDS));
+ // The node remains running: stop() represents replacing just the reporter
on reload.
+ reporter.stop();
+ assertTrue(service.awaitTermination(5, TimeUnit.SECONDS));
+
+ assertTrue(interrupted.get());
+ ExecutionResult result = returnedResult.get();
+ if (recordedFailure != null) {
+ assertSame(recordedFailure, result.status);
+ } else {
+ assertEquals(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode(),
result.status.getCode());
+ if (recordedException != null) {
+ assertEquals(recordedException.getMessage(),
result.status.getMessage());
+ }
+ }
+ executedStatements(autoCreate ? 2 : 1);
+ assertEquals(
+ recordedFailure == null && recordedException == null,
result.failureCausedByInterruption);
+ assertEquals(recordedFailure != null || recordedException != null,
hasWarning());
+ }
+
+ private QueryExecution queryExecutionWithDelayedStatus(
+ TSStatus recordedFailure, Throwable recordedException, CountDownLatch
waitingForStatus)
+ throws Exception {
+ MPPQueryContext context = mock(MPPQueryContext.class);
+ when(context.getQueryId()).thenReturn(new QueryId("metric_reload"));
+ IPlanner planner = mock(IPlanner.class);
+ when(planner.analyze(context)).thenReturn(mock(IAnalysis.class));
+ ExecutorService notifications = mock(ExecutorService.class);
+ QueryExecution execution = new QueryExecution(planner, context,
notifications);
+ QueryStateMachine stateMachine = stateMachine(execution);
+ if (recordedFailure != null) {
+ stateMachine.transitionToFailed(recordedFailure);
+ } else if (recordedException != null) {
+ stateMachine.transitionToFailed(recordedException);
+ }
+ // Delay listener delivery as with a busy notification executor. The real
getStatus() waits
+ // until shutdownNow() interrupts it, even when the state machine already
holds a failure.
+ doAnswer(
+ invocation -> {
+ waitingForStatus.countDown();
+ return null;
+ })
+ .when(notifications)
+ .execute(any(Runnable.class));
+ return execution;
+ }
+
+ private QueryStateMachine stateMachine(QueryExecution execution) throws
Exception {
+ Field stateMachineField =
QueryExecution.class.getDeclaredField("stateMachine");
+ stateMachineField.setAccessible(true);
+ return (QueryStateMachine) stateMachineField.get(execution);
+ }
+
+ private IoTDBInternalLocalReporter newReporter(ScheduledExecutorService
executor) {
+ return new IoTDBInternalLocalReporter(
+ executor, coordinator, sessionInfo, partitionFetcher, schemaFetcher);
+ }
+
+ private OngoingStubbing<ExecutionResult> execution() {
+ return when(
+ coordinator.executeForTreeModel(
+ any(Statement.class),
+ anyLong(),
+ same(sessionInfo),
+ eq(""),
+ same(partitionFetcher),
+ same(schemaFetcher)));
+ }
+
+ private ExecutionResult result(TSStatusCode code) {
+ return new ExecutionResult(new QueryId("metric_test"), new
TSStatus(code.getStatusCode()));
+ }
+
+ private void writeMetric() {
+ reporter.writeMetricToIoTDB(
+ Collections.singletonMap("value", 1L), "root.__system.metric.test",
1L);
+ }
+
+ private List<Statement> executedStatements(int count) {
+ ArgumentCaptor<Statement> statement =
ArgumentCaptor.forClass(Statement.class);
+ verify(coordinator, times(count))
+ .executeForTreeModel(
+ statement.capture(),
+ anyLong(),
+ same(sessionInfo),
+ eq(""),
+ same(partitionFetcher),
+ same(schemaFetcher));
+ return statement.getAllValues();
+ }
+
+ private boolean hasWarning() {
+ return appender.list.stream().anyMatch(event ->
event.getLevel().isGreaterOrEqual(Level.WARN));
+ }
+}
