diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java index 0d9ca912571a..cf359bec5f55 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java @@ -165,6 +165,11 @@ public ConfigPhysicalPlanType getType() { return this.type; } + /** + * Serializes this plan, including its type discriminator and implementation-specific payload. + * + * @return a buffer positioned at the beginning of the serialized plan + */ @Override public ByteBuffer serializeToByteBuffer() { try (final PublicBAOS byteArrayOutputStream = new PublicBAOS(); @@ -189,6 +194,13 @@ public int getSerializedSize() throws IOException { public static class Factory { + /** + * Deserializes a plan from the buffer using the encoded type discriminator. + * + * @param buffer the buffer containing one serialized plan + * @return the deserialized plan + * @throws IOException if the encoded plan type or payload cannot be read + */ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOException { final short planType = buffer.getShort(); final ConfigPhysicalPlanType configPhysicalPlanType = diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanVisitor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanVisitor.java index 53e3c4cd37df..ba4948600541 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanVisitor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanVisitor.java @@ -211,7 +211,13 @@ public R process(final ConfigPhysicalPlan plan, final C context) { } } - /** Top Level Description */ + /** + * Dispatches ConfigPhysicalPlan instances to type-specific visitor methods. + * + *

When a new plan type is introduced, the top-level dispatch and the corresponding visitor + * method must be updated together. Default visitor methods should document whether they delegate + * to the top-level handler, return a default value, or reject the plan. + */ public abstract R visitPlan(final ConfigPhysicalPlan plan, final C context); public R visitCreateDatabase(final DatabaseSchemaPlan createDatabasePlan, final C context) { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java index 790f09c08b9f..e085b4ae7906 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java @@ -84,6 +84,13 @@ public String getClusterId() { return clusterInfo.getClusterId(); } + /** + * Waits up to the specified time for the cluster ID to become available. + * + * @param maxWaitTime maximum wait time in milliseconds + * @return the cluster ID, or null if it is unavailable after the timeout or the wait is + * interrupted + */ public String getClusterIdWithRetry(long maxWaitTime) { long startTime = System.currentTimeMillis(); while (clusterInfo.getClusterId() == null @@ -109,7 +116,12 @@ private void generateClusterId() { } } - // TODO: Parallel test ConfigNode and DataNode + /** + * Tests connectivity from this ConfigNode to all registered ConfigNodes and DataNodes and + * aggregates the results. + * + * @return aggregated connection-test results + */ public TTestConnectionResp submitTestConnectionTaskToEveryNode() { TTestConnectionResp resp = new TTestConnectionResp(); resp.resultList = new ArrayList<>(); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java index f86681f2bbec..af3e6a2cf3e3 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java @@ -310,7 +310,10 @@ import static org.apache.iotdb.commons.conf.IoTDBConstant.ONE_LEVEL_PATH_WILDCARD; import static org.apache.iotdb.commons.schema.table.Audit.TREE_MODEL_AUDIT_DATABASE; -/** Entry of all management, AssignPartitionManager, AssignRegionManager. */ +/** + * Coordinates ConfigNode control-plane managers and routes requests related to nodes, schema, + * partitions, procedures, load management, plugins, quotas, TTL, and subscriptions. + */ public class ConfigManager implements IManager { private static final Logger LOGGER = LoggerFactory.getLogger(ConfigManager.class); @@ -318,7 +321,7 @@ public class ConfigManager implements IManager { private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf(); private static final CommonConfig COMMON_CONF = CommonDescriptor.getInstance().getConfig(); - /** Manage PartitionTable read/write requests through the ConsensusLayer. */ + /** Manages replicated ConfigRegion plans and reads through the consensus layer. */ private final AtomicReference consensusManager = new AtomicReference<>(); /** Manage cluster-level info */ diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java index bbb807c51d00..14591a8da763 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java @@ -50,9 +50,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -/** - * EventService periodically check statistics and broadcast corresponding change event if necessary. - */ +/** Periodically checks cluster events that require ConfigNode-side handling. */ public class EventService { private static final Logger LOGGER = LoggerFactory.getLogger(EventService.class); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java index ef732145f22d..8ea35e5acdff 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java @@ -60,8 +60,8 @@ import java.util.stream.Collectors; /** - * HeartbeatService periodically sending heartbeat requests from ConfigNode-leader to all other - * cluster Nodes. + * Periodically sends heartbeat requests from the ConfigNode leader to registered ConfigNodes, + * DataNodes, and AINodes, and updates the corresponding load information. */ public class HeartbeatService { @@ -135,7 +135,7 @@ public void reloadHeartbeatInterval() { } } - /** loop body of the heartbeat thread. */ + /** Executes one leader heartbeat cycle. */ private void heartbeatLoopBody() { // The consensusManager of configManager may not be fully initialized at this time Optional.ofNullable(getConsensusManager()) @@ -300,7 +300,7 @@ private void pingRegisteredDataNodes( /** * Send heartbeat requests to all the Registered AINodes. * - * @param registeredAINodes DataNodes that registered in cluster + * @param registeredAINodes AINodes registered in the cluster */ private void pingRegisteredAINodes( TAIHeartbeatReq heartbeatReq, List registeredAINodes) { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java index 620276767f09..9f965a58df06 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java @@ -33,7 +33,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -/** StatisticsService periodically update load statistics for all load cache. */ +/** Periodically aggregates heartbeat samples into cluster load statistics. */ public class StatisticsService { private static final Logger LOGGER = LoggerFactory.getLogger(StatisticsService.class); @@ -44,7 +44,7 @@ public StatisticsService(LoadCache loadCache) { this.loadCache = loadCache; } - /** Load statistics executor service. */ + /** Guards load-statistics scheduling across lifecycle and leadership transitions. */ private final Object statisticsScheduleMonitor = new Object(); private Future currentLoadStatisticsFuture; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java index 64d4cf741e15..85715cf23c05 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java @@ -521,7 +521,8 @@ public TAINodeConfiguration getRegisteredAINode(int aiNodeId) { } /** - * Register AINode. Use synchronized to make sure + * Serialize AINode registration so concurrent requests cannot violate the single-AINode + * registration constraint. * * @param req TAINodeRegisterReq * @return AINodeConfigurationDataSet. The {@link TSStatus} will be set to {@link @@ -541,7 +542,7 @@ public synchronized DataSet registerAINode(TAINodeRegisterReq req) { int aiNodeId = nodeInfo.generateNextNodeId(); getLoadManager().getLoadCache().createNodeHeartbeatCache(NodeType.AINode, aiNodeId); RegisterAINodePlan registerAINodePlan = new RegisterAINodePlan(req.getAiNodeConfiguration()); - // Register new DataNode + // Register new AINode registerAINodePlan.getAINodeConfiguration().getLocation().setAiNodeId(aiNodeId); try { getConsensusManager().write(registerAINodePlan); @@ -549,7 +550,7 @@ public synchronized DataSet registerAINode(TAINodeRegisterReq req) { LOGGER.warn(CONSENSUS_WRITE_ERROR, e); } - // update datanode's versionInfo + // update AINode's versionInfo UpdateVersionInfoPlan updateVersionInfoPlan = new UpdateVersionInfoPlan(req.getVersionInfo(), aiNodeId); try { @@ -655,7 +656,7 @@ public DataNodeConfigurationResp getDataNodeConfiguration(GetDataNodeConfigurati } /** - * Only leader use this interface. + * Called only on the ConfigNode leader. * * @return The number of registered Nodes */ @@ -664,7 +665,7 @@ public int getRegisteredNodeCount() { } /** - * Only leader use this interface. + * Called only on the ConfigNode leader. * * @return The number of registered DataNodes */ @@ -673,7 +674,7 @@ public int getRegisteredDataNodeCount() { } /** - * Only leader use this interface. + * Called only on the ConfigNode leader. * * @return All registered DataNodes */ @@ -682,7 +683,7 @@ public List getRegisteredDataNodes() { } /** - * Only leader use this interface. + * Called only on the ConfigNode leader. * *

Notice: The result will be an empty TDataNodeConfiguration if the specified DataNode doesn't * register diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java index 28655923cca9..903964c855b3 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java @@ -151,8 +151,7 @@ public class PartitionManager { // Monitor for leadership change private final Object scheduleMonitor = new Object(); - /** Region cleaner. */ - // Try to delete Regions in every 10s + /** Period, in seconds, at which the region maintainer performs maintenance and cleanup. */ private static final int REGION_MAINTAINER_WORK_INTERVAL = 10; private final ScheduledExecutorService regionMaintainer; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java index 12f6a2baaa18..f7c5e055d92a 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java @@ -102,8 +102,8 @@ import static org.apache.iotdb.commons.schema.SchemaConstant.ROOT; import static org.apache.iotdb.commons.schema.SchemaConstant.TABLE_MNODE_TYPE; -// Since the ConfigMTree is all stored in memory, thus it is not restricted to manage MNode through -// MTreeStore. +// ConfigMTree stores nodes in memory. ConfigMTreeStore provides in-memory node access and traversal +// helpers. public class ConfigMTree { private static final String TABLE_ERROR_MSG = @@ -136,7 +136,7 @@ public void clear() { // region database Management /** - * CREATE DATABASE. Make sure check seriesPath before setting database + * Create the database after validating the path; intermediate nodes are created when necessary. * * @param path path */ diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java index 74141fa73f7f..5506620f0155 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java @@ -90,7 +90,7 @@ public final boolean hasLock() { * The code to undo what was done by the execute() code. It is called when the procedure or one of * the sub-procedures failed or an abort was requested. It should cleanup all the resources * created by the execute() call. The implementation must be idempotent since rollback() may be - * called multiple time in case of machine failure in the middle of the execution. + * called multiple times in case of machine failure in the middle of the execution. * * @param env the environment passed to the ProcedureExecutor * @throws IOException temporary failure, the rollback will retry later @@ -277,7 +277,7 @@ final void releaseExecution() { } /** - * Internal method called by the ProcedureExecutor that starts the user-level code execute(). + * Internal method called by the ProcedureExecutor that invokes the user-level execute(). * * @param env execute environment * @return sub procedures @@ -292,7 +292,7 @@ protected Procedure[] doExecute(Env env) throws InterruptedException { } /** - * Internal method called by the ProcedureExecutor that starts the user-level code rollback(). + * Internal method called by the ProcedureExecutor that invokes the user-level rollback(). * * @param env execute environment * @throws IOException ioe @@ -336,7 +336,7 @@ public final ProcedureLockState doAcquireLock(Env env, IProcedureStore store) { } /** - * Presist lock state of the procedure + * Persist the procedure's lock state. * * @param env environment * @param store ProcedureStore @@ -616,7 +616,7 @@ public boolean isLockedWhenLoading() { // Runtime state, updated every operation by the ProcedureExecutor // // There is always 1 thread at the time operating on the state of the procedure. - // The ProcedureExecutor may check and set states, or some Procecedure may + // The ProcedureExecutor may check and set states, or some Procedure may // update its own state. but no concurrent updates. we use synchronized here // just because the procedure can get scheduled on different executor threads on each step. // ============================================================================================== diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java index cf49a3916bf2..0ae69abf55cc 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNode.java @@ -156,10 +156,17 @@ protected void remove(Set nodeIds) throws IoTDBException { ConfigNodeMessages.THE_REMOVE_CONFIGNODE_SCRIPT_HAS_BEEN_DEPRECATED_PLEASE_CONNECT_TO, -1); } + /** + * Starts the ConfigNode services in dependency order. + * + *

The method must preserve the distinction between seed and non-seed ConfigNodes and must + * start the RPC service only after the local services required to handle requests are ready. + */ public void active() { LOGGER.info(ConfigNodeMessages.ACTIVATING, ConfigNodeConstant.GLOBAL_NAME); try { + // Process pid file, register deleteOnExit processPid(); // Add shutdown hook addShutDownHook(); @@ -503,6 +510,12 @@ public void deactivate() throws IOException { LOGGER.info(ConfigNodeMessages.IS_DEACTIVATED, ConfigNodeConstant.GLOBAL_NAME); } + /** + * Stops ConfigNode services and releases their resources in reverse dependency order. + * + *

The operation should be safe to invoke during partial startup and should not leave + * background scheduling, RPC, or consensus resources running. + */ public void stop() { try { deactivate(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java index 051dcbb27719..81ecf02f31f4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java @@ -55,6 +55,13 @@ import java.util.List; import java.util.function.Supplier; +/** + * Applies replicated write and query fragment operations to one DataRegion and exposes its snapshot + * and region-resource lifecycle to the consensus layer. + * + *

Write-process rejection is retried here to preserve the atomicity expected by the consensus + * apply path; other statuses are delegated to the consensus retry mechanism. + */ public class DataRegionStateMachine extends BaseStateMachine { private static final Logger logger = LoggerFactory.getLogger(DataRegionStateMachine.class); @@ -74,7 +81,8 @@ public DataRegionStateMachine(DataRegion region) { @Override public void start() { - // do nothing + // The consensus implementation owns the start lifecycle; this state machine has no additional + // start action. } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java index 9396d8cc3543..757b8e8714d0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java @@ -46,6 +46,10 @@ import java.util.List; import java.util.Objects; +/** + * Applies replicated schema operations to one SchemaRegion and coordinates schema-region snapshot, + * Pipe-leader, and attribute-security lifecycle callbacks. + */ public class SchemaRegionStateMachine extends BaseStateMachine { private static final Logger logger = LoggerFactory.getLogger(SchemaRegionStateMachine.class); @@ -60,7 +64,8 @@ public SchemaRegionStateMachine(final ISchemaRegion schemaRegion) { @Override public void start() { - // Do nothing + // The consensus implementation owns the start lifecycle; this state machine has no additional + // start action. } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java index be0a00bddf82..76c42642938a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java @@ -254,6 +254,13 @@ import static org.apache.iotdb.rpc.RpcUtils.TIME_PRECISION; import static org.apache.iotdb.rpc.TSStatusCode.QUERY_WAS_KILLED; +/** + * Implements the client-facing RPC surface for sessions, SQL execution, writes, query results, + * metadata operations, authentication, and resource control. + * + *

This class adapts protocol requests to the Coordinator and DataNode managers; it must preserve + * session ownership, query cleanup, authorization, timeout, and status-conversion semantics. + */ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { private static final Logger LOGGER = LoggerFactory.getLogger(ClientRPCServiceImpl.class); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java index 34ef951296ce..ac41fe5a7793 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java @@ -410,6 +410,13 @@ import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onIoTDBException; import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onQueryException; +/** + * Implements the internal DataNode RPC surface used for fragment execution, region lifecycle, + * schema/data operations, load, consensus-related coordination, and cluster maintenance. + * + *

Internal requests may arrive during startup, shutdown, migration, or recovery, so methods must + * document their readiness checks and idempotency behavior where it is not obvious. + */ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface { private static final Logger LOGGER = LoggerFactory.getLogger(DataNodeInternalRPCServiceImpl.class); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java index ceba3880122a..5b6a5f178ad2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java @@ -88,6 +88,13 @@ import static org.apache.iotdb.db.queryengine.metric.DataExchangeCountMetricSet.ON_ACKNOWLEDGE_DATA_BLOCK_NUM_SERVER; import static org.apache.iotdb.db.queryengine.metric.DataExchangeCountMetricSet.SEND_NEW_DATA_BLOCK_NUM_SERVER; +/** + * Manages local and remote source/sink handles used to exchange TsBlocks between MPP fragments. + * + *

The manager processes data-block fetch, acknowledgement, close, and end-of-stream events. Late + * events are expected after downstream cancellation and must be ignored without leaking handles or + * corrupting completion state. + */ public class MPPDataExchangeManager implements IMPPDataExchangeManager { private static final Logger LOGGER = LoggerFactory.getLogger(MPPDataExchangeManager.class); @@ -299,10 +306,11 @@ public void onNewDataBlockEvent(TNewDataBlockEvent e) throws TException { : (SourceHandle) sourceHandleMap.get(e.getTargetPlanNodeId()); if (sourceHandle == null || sourceHandle.isAborted() || sourceHandle.isFinished()) { - // In some scenario, when the SourceHandle sends the data block ACK event, its upstream - // may - // have already been stopped. For example, in the read whit LimitOperator, the downstream - // FragmentInstance may be finished, although the upstream is still working. + // A downstream fragment may finish early, for example when a LimitOperator has produced + // enough + // rows, while its upstream fragment is still sending events. Ignore late events for the + // finished + // or aborted SourceHandle. if (LOGGER.isDebugEnabled()) { LOGGER.debug( DataNodeQueryMessages diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java index fa6a45e83749..103f338e7c34 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java @@ -73,6 +73,13 @@ import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceExecution.createFragmentInstanceExecution; import static org.apache.iotdb.rpc.TSStatusCode.TOO_MANY_CONCURRENT_QUERIES_ERROR; +/** + * Creates, tracks, executes, and cleans up fragment instances on this DataNode. + * + *

The manager owns fragment contexts and executions, constructs local pipeline drivers, rejects + * repeated dispatches, schedules timeout cleanup, and releases query-level resources when an + * instance reaches a terminal state. + */ @SuppressWarnings("squid:S6548") public class FragmentInstanceManager { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverScheduler.java index 0485de570fe5..c52dbf8187ff 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverScheduler.java @@ -67,7 +67,13 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; -/** The manager of fragment instances scheduling. */ +/** + * Schedules DataNode DriverTasks across ready, blocked, and timeout queues. + * + *

The scheduler enforces query and task capacity, accounts for CPU and memory quotas, moves + * blocked tasks back to the ready queue, and aborts tasks when their query or fragment instance is + * cancelled or timed out. + */ public class DriverScheduler implements IDriverScheduler, IService { private static final Logger logger = LoggerFactory.getLogger(DriverScheduler.class); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java index 09a35803595f..9bfd5204b0f1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java @@ -194,9 +194,10 @@ import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfCharArray; /** - * The coordinator for MPP. It manages all the queries which are executed in current Node. And it - * will be responsible for the lifecycle of a query. A query request will be represented as a - * QueryExecution. + * Coordinates the lifecycle of queries executed on this DataNode. + * + *

The coordinator creates query contexts, analyzes and plans statements, dispatches fragment + * instances, tracks query state, handles retries, and releases query resources after completion. */ public class Coordinator { 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 40f5231a81fb..39fa4f4d7023 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 @@ -79,10 +79,11 @@ import static org.apache.iotdb.rpc.TSStatusCode.DATE_OUT_OF_RANGE; /** - * QueryExecution stores all the status of a query which is being prepared or running inside the MPP - * frame. It takes three main responsibilities: 1. Prepare a query. Transform a query from statement - * to DistributedQueryPlan with fragment instances. 2. Dispatch all the fragment instances to - * corresponding physical nodes. 3. Collect and monitor the progress/states of this query. + * Represents the lifecycle and execution state of one MPP query or write operation. + * + *

It analyzes the statement, builds logical and distributed plans, dispatches fragment + * instances, exposes results through the local or remote exchange layer, monitors state changes, + * retries eligible failures, and releases resources exactly once. */ public class QueryExecution implements IQueryExecution { private static final Logger LOGGER = LoggerFactory.getLogger(QueryExecution.class); @@ -101,11 +102,12 @@ public class QueryExecution implements IQueryExecution { private LogicalQueryPlan logicalPlan; private DistributedQueryPlan distributedPlan; - // The result of QueryExecution will be written to the MPPDataExchangeManager in current Node. - // We use this SourceHandle to fetch the TsBlock from it. + /** + * Result blocks are published to the local exchange manager and read through this source handle. + */ private ISourceHandle resultHandle; - // used for cleaning resultHandle up exactly once + /** Guards exactly-once cleanup of the result source handle. */ private final AtomicBoolean resultHandleCleanUp; private final AtomicBoolean stopped; @@ -199,7 +201,8 @@ private void startInternal() { return; } - // check timeout for query first + // Apply the timeout only to query operations. Write operations use the write-path retry and + // backpressure rules instead of a query execution deadline. checkTimeOutForQuery(); doLogicalPlan(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java index 5f2f72f43139..2c91a9e14086 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java @@ -171,7 +171,12 @@ public R process(StatementNode node, C context) { return node.accept(this, context); } - /** Top Level Description */ + /** + * Visits the root of the statement hierarchy. + * + *

Default visitor methods delegate to this method, allowing subclasses to handle only the + * statement types they need. + */ public abstract R visitNode(StatementNode node, C context); public R visitStatement(Statement statement, C context) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java index ad8042ec7030..c5e41e573be6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java @@ -73,7 +73,14 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -// manage all the schemaRegion in this dataNode +/** + * Owns the schema regions hosted by this DataNode and coordinates their loading, recovery, metrics, + * schema-resource management, and lifecycle operations. + * + *

The selected schema-engine mode determines whether schema state is memory-resident or cached. + * Schema metrics and shared schema resources must be initialized before schema regions, and cleared + * only after all schema regions have been cleared. + */ public class SchemaEngine { private static final Logger logger = LoggerFactory.getLogger(SchemaEngine.class); @@ -118,8 +125,9 @@ public void init() { initSchemaEngineStatistics(); SchemaResourceManager.initSchemaResource(schemaEngineStatistics); - // CachedSchemaEngineMetric depend on CacheMemoryManager, so it should be initialized after - // CacheMemoryManager + + // Cached schema metrics depend on CacheMemoryManager, so initialize them only after the cache + // memory manager is ready. schemaMetricManager = new SchemaMetricManager(schemaEngineStatistics); initSchemaRegion(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java index 5ce282db7a70..8f0c286f8cd8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java @@ -156,6 +156,14 @@ import static org.apache.iotdb.commons.utils.StatusUtils.retrieveExitStatusCode; import static org.apache.iotdb.db.conf.IoTDBStartCheck.PROPERTIES_FILE_NAME; +/** + * The process-level service for a DataNode. + * + *

DataNode registers with the ConfigNode cluster, restores local schema and data regions, starts + * consensus and query services, and exposes client and internal RPC endpoints. Startup order is + * significant because query, write, and region-management services must not serve requests before + * local recovery and runtime configuration have completed. + */ public class DataNode extends ServerCommandLine implements DataNodeMBean { private static final Logger logger = LoggerFactory.getLogger(DataNode.class); @@ -252,13 +260,21 @@ public static void main(final String[] args) { } } + /** + * Starts the DataNode by preparing local state, synchronizing cluster configuration, registering + * or restarting the node, recovering regions, and starting the remaining services in dependency + * order. + * + *

The first-start and restart paths intentionally perform different registration and security + * checks. + */ @Override protected void start() { logger.info(DataNodeMiscMessages.STARTING_DATANODE); boolean isFirstStart; try { IoTDBDescriptor.getInstance().getMemoryConfig().activateAutoResizingBufferMemoryControl(); - // Check if this DataNode is start for the first time and do other pre-checks + // Check whether this is the first DataNode startup and run the remaining startup checks. isFirstStart = prepareDataNode(); if (isFirstStart) { @@ -1368,6 +1384,13 @@ public void deleteDataNodeSystemProperties() { DataNodeSystemPropertiesHandler.getInstance().delete(); } + /** + * Stops DataNode services and releases resources in an order that preserves WAL, TsFile, + * consensus, query, and RPC shutdown dependencies. + * + *

The method must remain safe during partial startup because startup failures invoke it as + * cleanup. + */ public void stop() { stopTriggerRelatedServices(); registerManager.deregisterAll(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java index 6dd082c0efa4..0956deed827d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java @@ -56,6 +56,14 @@ import java.util.Map; +/** + * Performs an orderly DataNode shutdown. + * + *

The hook first prevents new writes and drains write-related resources, then closes or + * snapshots storage according to the configured consensus protocol, persists Pipe progress, stops + * DataNode services, reports shutdown to the ConfigNode leader, and finally releases the directory + * lock. + */ public class DataNodeShutdownHook extends Thread { private static final Logger logger = LoggerFactory.getLogger(DataNodeShutdownHook.class); @@ -134,9 +142,8 @@ public void run() { // We did this work because the RatisConsensus recovery mechanism is different from other // consensus algorithms, which will replace the underlying storage engine based on its - // own - // latest snapshot, while other consensus algorithms will not. This judgement ensures that - // compaction work is not discarded even if there are frequent restarts + // own latest snapshot, while other consensus algorithms will not. This judgement ensures + // that compaction work is not discarded even if there are frequent restarts if (IoTDBDescriptor.getInstance() .getConfig() .getDataRegionConsensusProtocolClass() @@ -189,9 +196,8 @@ public void run() { // set encryption key to 16-byte zero. TSFileDescriptor.getInstance().getConfig().setEncryptKey(new byte[16]); - // Actually stop all services started by the DataNode. - // If we don't call this, services like the RestService are not stopped and I can't re-start - // it. + // Stop every service started by DataNode. Otherwise services such as RestService may retain + // resources and prevent a subsequent restart in the same JVM. DataNode.getInstance().stop(); // Set and report shutdown to cluster ConfigNode-leader diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 1ac6c15488b1..cf7ea4040425 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -149,11 +149,13 @@ public class StorageEngine implements IService { private final ConcurrentHashMap deletingDataRegionMap = new ConcurrentHashMap<>(); - /** number of ready data region */ + /** Number of data regions that have completed the initial local recovery phase. */ private AtomicInteger readyDataRegionNum; + /** True when recovered data regions can serve ordinary reads and writes. */ private final AtomicBoolean isReadyForReadAndWrite = new AtomicBoolean(); + /** True when non-read/write services may proceed with their initialization. */ private final AtomicBoolean isReadyForNonReadWriteFunctions = new AtomicBoolean(); private ScheduledExecutorService seqMemtableTimedFlushCheckThread; @@ -317,6 +319,10 @@ public Map> getLocalDataRegionInfo() { return localDataRegionInfo; } + /** + * Initializes storage directories, recovers data regions and WALs, starts timed services, and + * waits until the read/write readiness barrier is satisfied. + */ @Override public void start() throws StartupException { recoverDataRegionNum = 0; @@ -801,8 +807,8 @@ private void makeSureNoOldRegion(DataRegionId regionId) { } } - // When registering a new region, the coordinator needs to register the corresponding region with - // the local storage before adding the corresponding consensusGroup to the consensus layer + // Create the local DataRegion before the coordinator adds the corresponding consensus group, so + // consensus requests cannot arrive before the region and its lock are available. public void createDataRegion(DataRegionId regionId, String databaseName) throws DataRegionException { makeSureNoOldRegion(regionId); @@ -892,10 +898,10 @@ public TSStatus deleteDataRegion(DataRegionId regionId) { } /** - * run the runnable if the region is absent. if the region is present, do nothing. + * Runs the action only when the specified DataRegion is absent. * - *

we don't use computeIfAbsent because we don't want to create a new region if the region is - * absent, we just want to run the runnable in a synchronized way. + *

Uses computeIfAbsent to serialize the absence check and action; returning null keeps the map + * unchanged. * * @return true if the region is absent and the runnable is run. false if the region is present. */ @@ -912,10 +918,10 @@ public boolean runIfAbsent(DataRegionId regionId, Runnable runnable) { } /** - * run the consumer if the region is present. if the region is absent, do nothing. + * Run the consumer if the region is present. if the region is absent, do nothing. * - *

we don't use computeIfPresent because we don't want to remove the region if the consumer - * returns null, we just want to run the consumer in a synchronized way. + *

Uses computeIfPresent to serialize consumer invocation and returns the existing region so it + * remains in the map. * * @return true if the region is present and the consumer is run. false if the region is absent. */ @@ -947,7 +953,12 @@ public int getDataRegionNumber() { return dataRegionMap.size(); } - /** This method is not thread-safe */ + /** + * Replaces a local DataRegion while loading a snapshot. + * + *

This method is not thread-safe and may be called only while external region access is + * quiesced by the snapshot-loading protocol. + */ public DataRegion setDataRegionForSnapshotLoad( DataRegionId regionId, Supplier newRegionSupplier) { if (dataRegionMap.containsKey(regionId)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 6b14b43edf5f..17f1ea531881 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -289,7 +289,7 @@ public class DataRegion implements IDataRegionForQuery { /** Data region has been deleted or not. */ private volatile boolean deleted = false; - /** closeStorageGroupCondition is used to wait for all currently closing TsFiles to be done. */ + /** Condition used to wait until all TsFileProcessors currently closing have finished. */ private final Object closeStorageGroupCondition = new Object(); /** time partition id in the database -> {@link TsFileProcessor} for this time partition. */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java index 9024b0995f2f..623a43f2dd92 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java @@ -49,6 +49,12 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; +/** + * Schedules compaction-selection and TTL-check tasks for all local DataRegions. + * + *

Schedule tasks are protected by the manager lock. Configuration changes are applied only after + * active tasks have stopped, and repair tasks temporarily prevent incompatible schedule changes. + */ public class CompactionScheduleTaskManager implements IService { private int compactionSelectorNum = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java index 169d10137d8b..91a1c4c5833a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java @@ -53,7 +53,12 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; -/** CompactionMergeTaskPoolManager provides a ThreadPool tPro queue and run all compaction tasks. */ +/** + * Owns the worker pools and candidate queue used to execute DataRegion compaction tasks. + * + *

The manager tracks task futures by database and DataRegion, applies compaction rate limits, + * and coordinates graceful or immediate shutdown of compaction workers. + */ @SuppressWarnings("squid:S6548") public class CompactionTaskManager implements IService { private static final Logger logger = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index de1a5a09b7dd..67103c9b13f3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -127,10 +127,17 @@ import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.FLUSHING_MEMTABLE; import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.WORKING_MEMTABLE; +/** + * Manages one writable TsFile, its working and flushing MemTables, WAL entries, flush lifecycle, + * and resource metadata. + * + *

The flush/query lock coordinates reads, asynchronous flush, synchronous close, deletion, and + * resource publication. A processor is closed only after its pending MemTables have been flushed. + */ @SuppressWarnings("java:S1135") // ignore todos public class TsFileProcessor { - /** Logger fot this class. */ + /** Logger for this class. */ private static final Logger logger = LoggerFactory.getLogger(TsFileProcessor.class); private static final int NUM_MEM_TO_ESTIMATE = 3; @@ -167,12 +174,12 @@ public class TsFileProcessor { */ private volatile boolean managedByFlushManager; - /** A lock to mutual exclude read and read */ + /** Read/write lock coordinating query access with flush, close, and deletion operations. */ private final ReadWriteLock flushQueryLock = new ReentrantReadWriteLock(); /** - * It is set by the StorageGroupProcessor and checked by flush threads. (If shouldClose == true - * and its flushingMemTables are all flushed, then the flush thread will close this file.) + * Set by DataRegion when this processor must close after all MemTables currently being flushed + * have completed. */ private volatile boolean shouldClose; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java index 8fafee304b2b..576858b51a4d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java @@ -56,18 +56,28 @@ import static org.apache.iotdb.commons.conf.IoTDBConstant.FILE_NAME_SEPARATOR; -/** This class is used to manage and allocate wal nodes. */ +/** + * Allocates, tracks, flushes, and removes write-ahead-log nodes used by DataRegion write paths. + * + *

The allocation and deletion behavior depends on the configured consensus protocol and WAL + * mode. Disabled WAL mode must remain a no-op for lifecycle and allocation operations. + */ public class WALManager implements IService { private static final Logger logger = LoggerFactory.getLogger(WALManager.class); private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); - // manage all wal nodes and decide how to allocate them + /** + * Allocates WAL nodes and applies the strategy selected for the configured consensus protocol. + */ private final NodeAllocationStrategy walNodesManager; - // single thread to delete old .wal files + + /** Single-thread scheduler that deletes expired WAL files. */ private ScheduledExecutorService walDeleteThread; - // total disk usage of wal files + + /** Aggregate disk usage of all WAL nodes. */ private final AtomicLong totalDiskUsage = new AtomicLong(); - // total number of wal files + + /** Aggregate number of WAL files across all WAL nodes. */ private final AtomicLong totalFileNum = new AtomicLong(); private WALManager() { @@ -101,7 +111,7 @@ public IWALNode applyForWALNode(String applicantUniqueId) { return walNodesManager.applyForWALNode(applicantUniqueId); } - /** WAL node will be registered only when using iot series consensus protocol. */ + /** Registers a WAL node only for IoTConsensus and IoTConsensusV2. */ public void registerWALNode( String applicantUniqueId, String logDirectory, long startFileVersion, long startSearchIndex) { if (config.getWalMode() == WALMode.DISABLE @@ -117,7 +127,7 @@ public void registerWALNode( WritingMetrics.getInstance().createWALNodeInfoMetrics(applicantUniqueId); } - /** WAL node will be deleted only when using iot series consensus protocol. */ + /** Deletes a WAL node only for IoTConsensus and IoTConsensusV2. */ public void deleteWALNode(String applicantUniqueId) { if (config.getWalMode() == WALMode.DISABLE || (!config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java index d67cd88a4c48..496b812f8c5b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java @@ -71,8 +71,10 @@ import static org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode.DEFAULT_SEARCH_INDEX; /** - * This buffer guarantees the concurrent safety and uses double buffers mechanism to accelerate - * writes and avoid waiting for buffer syncing to disk. + * Buffers WAL entries with a lock-protected working, syncing, and idle buffer rotation. + * + *

The rotation allows serialization to continue while another buffer is written to disk. All + * buffer-state transitions must follow {@code buffersLock} and its conditions. */ public class WALBuffer extends AbstractWALBuffer { private static final Logger logger = LoggerFactory.getLogger(WALBuffer.class); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java index 54a45626a3d6..846d518f3cd5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java @@ -51,7 +51,11 @@ import static org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils.getTsFileRelativePath; -/** First set allVsgScannedLatch, then call recover method. */ +/** + * Coordinates WAL recovery after every DataRegion has scanned its local unsealed TsFiles. + * + *

Callers must install the all-data-region-scanned latch before invoking {@link #recover()}. + */ public class WALRecoverManager { private static final Logger logger = LoggerFactory.getLogger(WALRecoverManager.class); private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgent.java index 9118891a79ea..653b81c830d8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgent.java @@ -19,6 +19,12 @@ package org.apache.iotdb.db.subscription.agent; +/** + * Entry point for DataNode subscription receiver, runtime, consumer, broker, and topic agents. + * + *

The singleton exposes the agents that own subscription protocol handling and runtime state; + * lifecycle and ownership rules are implemented by those agents. + */ public class SubscriptionAgent { private final SubscriptionReceiverAgent receiverAgent; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java index b0eec505a5c4..ded035b1d546 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java @@ -103,6 +103,13 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; +/** + * Handles version-1 subscription requests, including handshake, heartbeat, subscribe, poll, commit, + * seek, unsubscribe, and consumer close operations. + * + *

Consumer state is shared across request threads and is fenced by consumer ownership. Poll, + * commit, seek, timeout, and exit paths must preserve in-flight request and progress invariants. + */ public class SubscriptionReceiverV1 implements SubscriptionReceiver { private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionReceiverV1.class);