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 c89034e8831d..35a2cf57f40e 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 @@ -1243,6 +1243,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 = @@ -3456,6 +3457,7 @@ private 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 a3307bd07cb4..a32008d51ad0 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 @@ -1223,6 +1223,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 = @@ -4107,6 +4108,7 @@ private 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 1bc3e2b0f324..2683fec32cf0 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; private final AtomicReference failureException = new AtomicReference<>(); - private final AtomicReference failureStatus = new AtomicReference<>(); + private final AtomicReference failureStatus = new AtomicReference<>(); public QueryStateMachine(QueryId queryId, ExecutorService executor) { this.queryState = @@ -90,7 +90,7 @@ public void transitionToDispatching() { } 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 void transitionToCanceled() { } 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 void transitionToFailed(Throwable throwable) { } 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 Throwable getFailureException() { } 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 TSStatus getFailureStatus() { 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 ac34cccb4373..ee5eb8623bf8 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 void run() { 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 void run() { 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 fb683db9962b..0f800376d33e 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 c00df6d8c730..2f46460635ad 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 5a0948146078..b8f753d70a01 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 ExecutionResult getStatus() { } 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 @@ private ExecutionResult getExecutionResult(QueryState state) { || 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 5e0aa02f5c0c..4c828c3808cb 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 void start() { 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 6a6211394555..dd9d39694e8f 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 boolean isSuccessful() { 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 24c42a0c16fc..4894a8bc106f 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 @@ private Future dispatchWrite(List 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 1630175738f9..ad13233dbac2 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.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.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 configNodeClientManager = ConfigNodeClientManager.getInstance(); @@ -125,13 +131,29 @@ public IoTDBInternalLocalReporter() { } } + 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 boolean start() { } @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 ReporterType getReporterType() { @Override protected void writeMetricToIoTDB(Map 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 valueMap, String prefix, long time) + private void writeMetric(Map 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 valueMap, String prefix, long time) throws IoTDBConnectionException, QueryProcessException, IllegalPathException { TSInsertRecordReq request = new TSInsertRecordReq(); List measurements = new ArrayList<>(); @@ -214,10 +292,8 @@ private TSStatus insertRecord(Map valueMap, String prefix, long 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 valueMap, String prefix) @@ -244,10 +320,14 @@ private void createTimeSeries(Map valueMap, String prefix) 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 ad7ab0947481..118ff7b5025b 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.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 void testTransitionToPendingRetry() { 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 000000000000..8f57718c665b --- /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 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 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 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 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 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 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 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 pollingQueue(CountDownLatch polling) { + return new L1PriorityQueue( + 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 959bd4b9d6ab..0c154850ca5a 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.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 void setup() { @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 000000000000..a94cefc54c00 --- /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 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.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 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 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 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 returnedResult = new AtomicReference<>(); + OngoingStubbing 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.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 returnedResult = new AtomicReference<>(); + AtomicBoolean interrupted = new AtomicBoolean(); + OngoingStubbing 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 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 executedStatements(int count) { + ArgumentCaptor 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)); + } +}