This is an automated email from the ASF dual-hosted git repository. JackieTien97 pushed a commit to branch codex/fix-shutdown-interruption-logs in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b35298945277f62f0dd560f42b2b053814055f85 Author: JackieTien97 <[email protected]> AuthorDate: Tue Oct 20 17:13:59 2026 +0800 Stop internal metric writes cleanly during shutdown --- .../metrics/IoTDBInternalLocalReporter.java | 150 ++++++-- .../metrics/IoTDBInternalLocalReporterTest.java | 378 +++++++++++++++++++++ 2 files changed, 494 insertions(+), 34 deletions(-) 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..6378476e103 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,29 +190,83 @@ 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 void writeMetric(Map<String, Object> valueMap, String prefix, long time) { + if (isStopping()) { + return; + } + try { + TSStatus result = insertRecord(valueMap, prefix, time); + if (result.getCode() == TSStatusCode.METADATA_ERROR.getStatusCode() && !isStopping()) { + createTimeSeries(valueMap, prefix); + if (isStopping()) { + return; + } + result = insertRecord(valueMap, prefix, time); + } + if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + if (isExpectedShutdownFailure(result)) { + LOGGER.debug(DataNodeMiscMessages.FAILED_UPDATE_METRIC_VALUE, result); + } else { + LOGGER.warn(DataNodeMiscMessages.FAILED_UPDATE_METRIC_VALUE, result); + } + } + } 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(TSStatus status) { + return isStopping() + && (Thread.currentThread().isInterrupted() + || (CommonDescriptor.getInstance().getConfig().isStopping() + && 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 TSStatus insertRecord(Map<String, Object> valueMap, String prefix, long time) @@ -215,7 +293,7 @@ public class IoTDBInternalLocalReporter extends IoTDBInternalReporter { InsertRowStatement s = StatementGenerator.createStatement(request); final long queryId = SESSION_MANAGER.requestQueryId(); ExecutionResult result = - COORDINATOR.executeForTreeModel( + coordinator.executeForTreeModel( s, queryId, sessionInfo, "", partitionFetcher, schemaFetcher); return result.status; } @@ -244,10 +322,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.status)) { + 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/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..718421b6dac --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporterTest.java @@ -0,0 +1,378 @@ +/* + * 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.TSStatus; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.plan.Coordinator; +import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; +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.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.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.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +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.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()); + assertFalse(hasWarning()); + } + + @Test + public void testStopInterruptsAutoCreateAndPreventsRetry() throws Exception { + // Exercise shutdownNow() on a real worker while schema creation is in progress. + service = Executors.newSingleThreadScheduledExecutor(); + reporter = newReporter(service); + CountDownLatch creating = new CountDownLatch(1); + execution() + .thenReturn(result(TSStatusCode.METADATA_ERROR)) + .thenAnswer( + invocation -> { + creating.countDown(); + try { + new CountDownLatch(1).await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return result(TSStatusCode.INTERNAL_SERVER_ERROR); + }); + + writeMetric(); + assertTrue(creating.await(5, TimeUnit.SECONDS)); + reporter.stop(); + assertTrue(service.awaitTermination(5, TimeUnit.SECONDS)); + + assertEquals(2, executedStatements(2).size()); + assertFalse(hasWarning()); + } + + @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 testInternalErrorWithoutShutdownStillWarns() { + execution() + .thenAnswer( + invocation -> { + Thread.currentThread().interrupt(); + return result(TSStatusCode.INTERNAL_SERVER_ERROR); + }); + + 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()); + } + + 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)); + } +}
