This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch fix_stop_schema_region_npe in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 126b14e5a9e7848cf50334562f3eaec5911f8155 Author: Tian Jiang <[email protected]> AuthorDate: Sat Oct 10 19:38:01 2026 +0800 Fix schema region NPE during DataNode shutdown --- .../iotdb/db/i18n/DataNodeQueryMessages.java | 2 + .../iotdb/db/i18n/DataNodeQueryMessages.java | 2 + .../execution/executor/RegionWriteExecutor.java | 12 ++- .../executor/RegionWriteExecutorTest.java | 98 ++++++++++++++++++++++ 4 files changed, 112 insertions(+), 2 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 c9e2a53506e..6276ba27806 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 @@ -20,6 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodeQueryMessages { + public static final String MESSAGE_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A = + "Schema region is unavailable. Please retry later."; public static final String EXCEPTION_FRAGMENT_INSTANCE_ARG_IS_ALREADY_ARG_B44984B4 = "Fragment instance %s is already %s"; 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 857031856b7..31d0682f531 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 @@ -20,6 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodeQueryMessages { + public static final String MESSAGE_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A = + "SchemaRegion 不可用，请稍后重试。"; public static final String EXCEPTION_FRAGMENT_INSTANCE_ARG_IS_ALREADY_ARG_B44984B4 = "Fragment instance %s 已处于 %s 状态"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java index 6fc521a7f72..fdf5207334a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java @@ -842,9 +842,9 @@ public class RegionWriteExecutor { } /** - * Check the quota before creating time series. + * Check region availability, data types and quota before creating time series. * - * @return null if the quota is not exceeded, otherwise return the execution result. + * @return null if all checks pass, otherwise return the execution result. */ private RegionExecutionResult checkQuotaAndTypeBeforeCreatingTimeSeries( final ISchemaRegion schemaRegion, @@ -852,6 +852,14 @@ public class RegionWriteExecutor { final int size, final List<String> measurements, final List<TSDataType> dataTypes) { + // Shutdown may clear SchemaEngine while the region's validation lock still exists. + // Reject the request as retryable before dereferencing the missing region. + if (schemaRegion == null) { + final String message = + DataNodeQueryMessages.MESSAGE_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A; + return RegionExecutionResult.create( + false, message, RpcUtils.getStatus(TSStatusCode.NO_AVAILABLE_REGION_GROUP, message)); + } for (int i = 0; i < measurements.size(); ++i) { if (dataTypes.get(i) == TSDataType.OBJECT) { final String errorStr = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java index ac3215d7027..2e99b7fab8c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutorTest.java @@ -22,14 +22,23 @@ package org.apache.iotdb.db.queryengine.execution.executor; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.commons.consensus.SchemaRegionId; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.IConsensus; import org.apache.iotdb.consensus.exception.ConsensusException; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.protocol.thrift.impl.DataNodeRegionManager; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.InternalCreateMultiTimeSeriesNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.InternalCreateTimeSeriesNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.MeasurementGroup; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedWritePlanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.schemaengine.SchemaEngine; +import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion; import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager; import org.apache.iotdb.db.trigger.executor.TriggerFireResult; import org.apache.iotdb.db.trigger.executor.TriggerFireVisitor; @@ -37,17 +46,106 @@ import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.trigger.api.enums.TriggerEvent; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.utils.Pair; import org.junit.Test; import org.mockito.Mockito; +import java.util.Collections; import java.util.concurrent.locks.ReentrantReadWriteLock; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; public class RegionWriteExecutorTest { + @Test + public void testInternalCreateTimeSeriesAfterSchemaRegionCleared() throws Exception { + // A shutdown can clear the region while its validation lock remains registered. + checkInternalCreateWithRegionAvailability(false, false); + } + + @Test + public void testInternalCreateMultiTimeSeriesAfterSchemaRegionCleared() throws Exception { + // Batch and Pipe auto-creation must return a retryable status for the same shutdown window. + checkInternalCreateWithRegionAvailability(true, false); + } + + @Test + public void testInternalCreateWithAvailableSchemaRegion() throws Exception { + // The guard must preserve quota checks and consensus writes for both live-region requests. + checkInternalCreateWithRegionAvailability(false, true); + checkInternalCreateWithRegionAvailability(true, true); + } + + private void checkInternalCreateWithRegionAvailability(boolean batch, boolean available) + throws Exception { + final String originalProtocol = + IoTDBDescriptor.getInstance().getConfig().getSchemaRegionConsensusProtocolClass(); + try { + for (String protocol : + new String[] {ConsensusFactory.SIMPLE_CONSENSUS, ConsensusFactory.RATIS_CONSENSUS}) { + IoTDBDescriptor.getInstance().getConfig().setSchemaRegionConsensusProtocolClass(protocol); + for (boolean fromPipe : new boolean[] {false, true}) { + final IConsensus consensus = Mockito.mock(IConsensus.class); + final DataNodeRegionManager regionManager = Mockito.mock(DataNodeRegionManager.class); + final SchemaEngine schemaEngine = Mockito.mock(SchemaEngine.class); + final SchemaRegionId regionId = new SchemaRegionId(1); + final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + Mockito.when(regionManager.getRegionLock(regionId)).thenReturn(lock); + final ISchemaRegion region = Mockito.mock(ISchemaRegion.class); + Mockito.when(schemaEngine.getSchemaRegion(regionId)) + .thenReturn(available ? region : null); + final RegionWriteExecutor executor = + new RegionWriteExecutor( + Mockito.mock(IConsensus.class), + consensus, + regionManager, + schemaEngine, + Mockito.mock(ClusterTemplateManager.class), + Mockito.mock(TriggerFireVisitor.class)); + final PartialPath device = new PartialPath("root.sg.d"); + final MeasurementGroup measurements = new MeasurementGroup(); + measurements.addMeasurement( + "s", TSDataType.INT64, TSEncoding.PLAIN, CompressionType.SNAPPY); + final WritePlanNode node = + batch + ? new InternalCreateMultiTimeSeriesNode( + new PlanNodeId("test"), + Collections.singletonMap(device, new Pair<>(false, measurements))) + : new InternalCreateTimeSeriesNode( + new PlanNodeId("test"), device, measurements, false); + final WritePlanNode request = fromPipe ? new PipeEnrichedWritePlanNode(node) : node; + Mockito.when(consensus.write(regionId, request)) + .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + + final RegionExecutionResult result = executor.execute(regionId, request); + assertEquals(available, result.isAccepted()); + assertEquals( + (available ? TSStatusCode.SUCCESS_STATUS : TSStatusCode.NO_AVAILABLE_REGION_GROUP) + .getStatusCode(), + result.getStatus().getCode()); + assertFalse(lock.isWriteLocked()); + assertEquals(0, lock.getReadLockCount()); + if (available) { + Mockito.verify(region).checkSchemaQuota(device, 1); + Mockito.verify(consensus).write(regionId, request); + } else { + Mockito.verifyZeroInteractions(consensus); + assertEquals(1, measurements.size()); + } + } + } + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setSchemaRegionConsensusProtocolClass(originalProtocol); + } + } + @Test public void testInsertRowNode() throws ConsensusException {
