This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch sync-generic-changes in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit a9567a078876eae9005ca246e8b3a69598e18e67 Author: shuwenwei <[email protected]> AuthorDate: Thu Sep 17 14:39:12 2026 +0800 [ConfigNode] Inject additional persistence and snapshot processors via ConfigManagerContext --- .../iotdb/confignode/manager/ConfigManager.java | 105 ++++++++++++--------- .../persistence/executor/ConfigPlanExecutor.java | 75 +++++++-------- 2 files changed, 94 insertions(+), 86 deletions(-) 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 f86681f2bbe..1c41618d90a 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 @@ -67,6 +67,7 @@ import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.commons.schema.tree.AlterTimeSeriesOperationType; import org.apache.iotdb.commons.schema.ttl.TTLCache; import org.apache.iotdb.commons.service.metric.MetricService; +import org.apache.iotdb.commons.snapshot.SnapshotProcessor; import org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper; import org.apache.iotdb.commons.subscription.meta.consumer.SubscriptionProgressSnapshot; import org.apache.iotdb.commons.utils.AuthUtils; @@ -377,58 +378,44 @@ public class ConfigManager implements IManager { public ConfigManager() throws IOException { // Build the persistence module - ClusterInfo clusterInfo = new ClusterInfo(); - NodeInfo nodeInfo = new NodeInfo(); - ClusterSchemaInfo clusterSchemaInfo = new ClusterSchemaInfo(); - PartitionInfo partitionInfo = new PartitionInfo(); - AuthorInfo authorInfo = createAuthorInfo(); - ProcedureInfo procedureInfo = new ProcedureInfo(this); - UDFInfo udfInfo = new UDFInfo(); - TriggerInfo triggerInfo = new TriggerInfo(); - CQInfo cqInfo = new CQInfo(); - ExternalServiceInfo externalServiceInfo = new ExternalServiceInfo(); - this.permissionManager = createPermissionManager(authorInfo); - PipeInfo pipeInfo = new PipeInfo(userName -> this.permissionManager.login4Pipe(userName, null)); - QuotaInfo quotaInfo = new QuotaInfo(); - TTLInfo ttlInfo = new TTLInfo(); - SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - + ConfigManagerContext context = createConfigManagerContext(); + context.clusterInfo = new ClusterInfo(); + context.nodeInfo = new NodeInfo(); + context.clusterSchemaInfo = new ClusterSchemaInfo(); + context.partitionInfo = new PartitionInfo(); + context.authorInfo = createAuthorInfo(); + context.procedureInfo = new ProcedureInfo(this); + context.udfInfo = new UDFInfo(); + context.triggerInfo = new TriggerInfo(); + context.cqInfo = new CQInfo(); + context.externalServiceInfo = new ExternalServiceInfo(); + this.permissionManager = createPermissionManager(context.authorInfo); + context.pipeInfo = new PipeInfo(userName -> this.permissionManager.login4Pipe(userName, null)); + context.quotaInfo = new QuotaInfo(); + context.ttlInfo = new TTLInfo(); + context.subscriptionInfo = new SubscriptionInfo(); + initAdditionalInfos(context); // Build state machine and executor - ConfigPlanExecutor executor = - new ConfigPlanExecutor( - clusterInfo, - nodeInfo, - clusterSchemaInfo, - partitionInfo, - authorInfo, - procedureInfo, - udfInfo, - triggerInfo, - cqInfo, - externalServiceInfo, - pipeInfo, - subscriptionInfo, - quotaInfo, - ttlInfo); + ConfigPlanExecutor executor = createConfigPlanExecutor(context); this.stateMachine = new ConfigRegionStateMachine(this, executor); // Build the manager module - this.clusterManager = new ClusterManager(this, clusterInfo); - setNodeManager(nodeInfo); + this.clusterManager = new ClusterManager(this, context.clusterInfo); + setNodeManager(context.nodeInfo); this.clusterSchemaManager = new ClusterSchemaManager( this, - clusterSchemaInfo, + context.clusterSchemaInfo, new ClusterSchemaQuotaStatistics( COMMON_CONF.getSeriesLimitThreshold(), COMMON_CONF.getDeviceLimitThreshold())); - this.partitionManager = new PartitionManager(this, partitionInfo); - this.procedureManager = createProcedureManager(procedureInfo); + this.partitionManager = new PartitionManager(this, context.partitionInfo); + this.procedureManager = createProcedureManager(context.procedureInfo); this.externalServiceManager = new ExternalServiceManager(this); - this.udfManager = new UDFManager(this, udfInfo); - this.triggerManager = new TriggerManager(this, triggerInfo); + this.udfManager = new UDFManager(this, context.udfInfo); + this.triggerManager = new TriggerManager(this, context.triggerInfo); this.cqManager = new CQManager(this); - this.pipeManager = new PipeManager(this, pipeInfo); - this.subscriptionManager = new SubscriptionManager(this, subscriptionInfo); + this.pipeManager = new PipeManager(this, context.pipeInfo); + this.subscriptionManager = new SubscriptionManager(this, context.subscriptionInfo); this.auditLogger = new CNAuditLogger(this); // 1. keep PipeManager initialization before LoadManager initialization, because @@ -438,8 +425,8 @@ public class ConfigManager implements IManager { setLoadManager(); this.retryFailedTasksThread = new RetryFailedTasksThread(this); - this.clusterQuotaManager = new ClusterQuotaManager(this, quotaInfo); - this.ttlManager = new TTLManager(this, ttlInfo); + this.clusterQuotaManager = new ClusterQuotaManager(this, context.quotaInfo); + this.ttlManager = new TTLManager(this, context.ttlInfo); } public void initConsensusManager() throws IOException { @@ -459,6 +446,16 @@ public class ConfigManager implements IManager { return new AuthorInfo(); } + protected ConfigManagerContext createConfigManagerContext() { + return new ConfigManagerContext(); + } + + protected void initAdditionalInfos(final ConfigManagerContext context) {} + + protected ConfigPlanExecutor createConfigPlanExecutor(final ConfigManagerContext context) { + return new ConfigPlanExecutor(context); + } + protected void setNodeManager(NodeInfo nodeInfo) { this.nodeManager = new NodeManager(this, nodeInfo); } @@ -3468,4 +3465,26 @@ public class ConfigManager implements IManager { public void setPermissionManager(final PermissionManager permissionManager) { this.permissionManager = permissionManager; } + + public static class ConfigManagerContext { + + public ClusterInfo clusterInfo; + public NodeInfo nodeInfo; + public ClusterSchemaInfo clusterSchemaInfo; + public PartitionInfo partitionInfo; + public AuthorInfo authorInfo; + public ProcedureInfo procedureInfo; + public UDFInfo udfInfo; + public TriggerInfo triggerInfo; + public CQInfo cqInfo; + public ExternalServiceInfo externalServiceInfo; + public PipeInfo pipeInfo; + public SubscriptionInfo subscriptionInfo; + public QuotaInfo quotaInfo; + public TTLInfo ttlInfo; + + public List<SnapshotProcessor> getAdditionalInfoList() { + return Collections.emptyList(); + } + } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java index 772f46baa31..58bf86d6858 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java @@ -156,6 +156,7 @@ import org.apache.iotdb.confignode.consensus.request.write.trigger.UpdateTrigger import org.apache.iotdb.confignode.consensus.response.partition.SchemaNodeManagementResp; import org.apache.iotdb.confignode.exception.physical.UnknownPhysicalPlanTypeException; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; +import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.manager.externalservice.ExternalServiceInfo; import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent; import org.apache.iotdb.confignode.persistence.ClusterInfo; @@ -228,65 +229,53 @@ public class ConfigPlanExecutor { private final TTLInfo ttlInfo; - public ConfigPlanExecutor( - ClusterInfo clusterInfo, - NodeInfo nodeInfo, - ClusterSchemaInfo clusterSchemaInfo, - PartitionInfo partitionInfo, - AuthorInfo authorInfo, - ProcedureInfo procedureInfo, - UDFInfo udfInfo, - TriggerInfo triggerInfo, - CQInfo cqInfo, - ExternalServiceInfo externalServiceInfo, - PipeInfo pipeInfo, - SubscriptionInfo subscriptionInfo, - QuotaInfo quotaInfo, - TTLInfo ttlInfo) { + public ConfigPlanExecutor(final ConfigManager.ConfigManagerContext context) { this.snapshotProcessorList = new ArrayList<>(); - this.clusterInfo = clusterInfo; - this.snapshotProcessorList.add(clusterInfo); + this.clusterInfo = context.clusterInfo; + this.snapshotProcessorList.add(context.clusterInfo); - this.nodeInfo = nodeInfo; - this.snapshotProcessorList.add(nodeInfo); + this.nodeInfo = context.nodeInfo; + this.snapshotProcessorList.add(context.nodeInfo); - this.clusterSchemaInfo = clusterSchemaInfo; - this.snapshotProcessorList.add(clusterSchemaInfo); + this.clusterSchemaInfo = context.clusterSchemaInfo; + this.snapshotProcessorList.add(context.clusterSchemaInfo); - this.partitionInfo = partitionInfo; - this.snapshotProcessorList.add(partitionInfo); + this.partitionInfo = context.partitionInfo; + this.snapshotProcessorList.add(context.partitionInfo); - this.authorInfo = authorInfo; - this.snapshotProcessorList.add(authorInfo); + this.authorInfo = context.authorInfo; + this.snapshotProcessorList.add(context.authorInfo); - this.triggerInfo = triggerInfo; - this.snapshotProcessorList.add(triggerInfo); + this.triggerInfo = context.triggerInfo; + this.snapshotProcessorList.add(context.triggerInfo); - this.udfInfo = udfInfo; - this.snapshotProcessorList.add(udfInfo); + this.udfInfo = context.udfInfo; + this.snapshotProcessorList.add(context.udfInfo); - this.cqInfo = cqInfo; - this.snapshotProcessorList.add(cqInfo); + this.cqInfo = context.cqInfo; + this.snapshotProcessorList.add(context.cqInfo); - this.externalServiceInfo = externalServiceInfo; - this.snapshotProcessorList.add(externalServiceInfo); + this.externalServiceInfo = context.externalServiceInfo; + this.snapshotProcessorList.add(context.externalServiceInfo); - this.pipeInfo = pipeInfo; - this.snapshotProcessorList.add(pipeInfo); + this.pipeInfo = context.pipeInfo; + this.snapshotProcessorList.add(context.pipeInfo); - this.subscriptionInfo = subscriptionInfo; - this.snapshotProcessorList.add(subscriptionInfo); + this.subscriptionInfo = context.subscriptionInfo; + this.snapshotProcessorList.add(context.subscriptionInfo); - this.procedureInfo = procedureInfo; - this.snapshotProcessorList.add(procedureInfo); + this.procedureInfo = context.procedureInfo; + this.snapshotProcessorList.add(context.procedureInfo); - this.quotaInfo = quotaInfo; - this.snapshotProcessorList.add(quotaInfo); + this.quotaInfo = context.quotaInfo; + this.snapshotProcessorList.add(context.quotaInfo); - this.ttlInfo = ttlInfo; - this.snapshotProcessorList.add(ttlInfo); + this.ttlInfo = context.ttlInfo; + this.snapshotProcessorList.add(context.ttlInfo); + + this.snapshotProcessorList.addAll(context.getAdditionalInfoList()); this.snapshotProcessorList.add(PipeConfigNodeAgent.runtime().listener()); }
