Optimized the hint for subscription (#17115)

Caideyipi · Caideyipi · commit 312e0477dc63 · 2026-02-02T10:16:20.000+08:00
* ams

* y

* fix
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigNodePluginAgent.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigNodePluginAgent.java
@@ -32,7 +32,7 @@ public PipeConfigNodePluginAgent(PipePluginMetaKeeper pipePluginMetaKeeper) {
   }
 
   @Override
-  protected PipeSourceConstructor createPipeExtractorConstructor(
+  protected PipeSourceConstructor createPipeSourceConstructor(
       PipePluginMetaKeeper pipePluginMetaKeeper) {
     return new PipeConfigRegionSourceConstructor();
   }
@@ -44,7 +44,7 @@ protected PipeProcessorConstructor createPipeProcessorConstructor(
   }
 
   @Override
-  protected PipeSinkConstructor createPipeConnectorConstructor(
+  protected PipeSinkConstructor createPipeSinkConstructor(
       PipePluginMetaKeeper pipePluginMetaKeeper) {
     return new PipeConfigRegionSinkConstructor();
   }
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/sink/protocol/IoTDBConfigRegionSink.java
@@ -183,9 +183,7 @@ private void doTransfer(final PipeConfigRegionWritePlanEvent pipeConfigRegionWri
           true);
     }
 
-    if (LOGGER.isDebugEnabled()) {
-      LOGGER.debug("Successfully transferred config event {}.", pipeConfigRegionWritePlanEvent);
-    }
+    LOGGER.info("Successfully transferred config event {}.", pipeConfigRegionWritePlanEvent);
   }
 
   private void doTransferWrapper(final PipeConfigRegionSnapshotEvent pipeConfigRegionSnapshotEvent)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeDataNodePluginAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/PipeDataNodePluginAgent.java
@@ -212,19 +212,18 @@ public void validate(
         pipeName, extractorAttributes, processorAttributes, connectorAttributes);
   }
 
-  public void validateExtractor(Map<String, String> extractorAttributes) throws Exception {
-    dataRegionAgent.validateExtractor(extractorAttributes);
-    schemaRegionAgent.validateExtractor(extractorAttributes);
+  public void validateSource(Map<String, String> sourceAttributes) throws Exception {
+    dataRegionAgent.validateSource(sourceAttributes);
+    schemaRegionAgent.validateSource(sourceAttributes);
   }
 
   public void validateProcessor(Map<String, String> processorAttributes) throws Exception {
     dataRegionAgent.validateProcessor(processorAttributes);
     schemaRegionAgent.validateProcessor(processorAttributes);
   }
 
-  public void validateConnector(String pipeName, Map<String, String> connectorAttributes)
-      throws Exception {
-    dataRegionAgent.validateConnector(pipeName, connectorAttributes);
-    schemaRegionAgent.validateConnector(pipeName, connectorAttributes);
+  public void validateSource(String pipeName, Map<String, String> sinkAttributes) throws Exception {
+    dataRegionAgent.validateSink(pipeName, sinkAttributes);
+    schemaRegionAgent.validateSink(pipeName, sinkAttributes);
   }
 }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionPluginAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionPluginAgent.java
@@ -33,7 +33,7 @@ public PipeDataRegionPluginAgent(DataNodePipePluginMetaKeeper pipePluginMetaKeep
   }
 
   @Override
-  protected PipeSourceConstructor createPipeExtractorConstructor(
+  protected PipeSourceConstructor createPipeSourceConstructor(
       PipePluginMetaKeeper pipePluginMetaKeeper) {
     return new PipeDataRegionSourceConstructor((DataNodePipePluginMetaKeeper) pipePluginMetaKeeper);
   }
@@ -46,7 +46,7 @@ protected PipeProcessorConstructor createPipeProcessorConstructor(
   }
 
   @Override
-  protected PipeSinkConstructor createPipeConnectorConstructor(
+  protected PipeSinkConstructor createPipeSinkConstructor(
       PipePluginMetaKeeper pipePluginMetaKeeper) {
     return new PipeDataRegionSinkConstructor((DataNodePipePluginMetaKeeper) pipePluginMetaKeeper);
   }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionPluginAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionPluginAgent.java
@@ -32,7 +32,7 @@ public PipeSchemaRegionPluginAgent(PipePluginMetaKeeper pipePluginMetaKeeper) {
   }
 
   @Override
-  protected PipeSourceConstructor createPipeExtractorConstructor(
+  protected PipeSourceConstructor createPipeSourceConstructor(
       PipePluginMetaKeeper pipePluginMetaKeeper) {
     return new PipeSchemaRegionSourceConstructor();
   }
@@ -44,7 +44,7 @@ protected PipeProcessorConstructor createPipeProcessorConstructor(
   }
 
   @Override
-  protected PipeSinkConstructor createPipeConnectorConstructor(
+  protected PipeSinkConstructor createPipeSinkConstructor(
       PipePluginMetaKeeper pipePluginMetaKeeper) {
     return new PipeSchemaRegionSinkConstructor();
   }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java
@@ -124,7 +124,7 @@ public PipeDataNodeTask build() {
             blendUserAndSystemParameters(pipeStaticMeta.getProcessorParameters()),
             regionId,
             sourceStage.getEventSupplier(),
-            sinkStage.getPipeConnectorPendingQueue(),
+            sinkStage.getPipeSinkPendingQueue(),
             PROCESSOR_EXECUTOR,
             pipeTaskMeta,
             pipeStaticMeta
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSinkStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSinkStage.java
@@ -34,7 +34,7 @@ public class PipeTaskSinkStage extends PipeTaskStage {
 
   protected final String pipeName;
   protected final long creationTime;
-  protected final PipeParameters pipeConnectorParameters;
+  protected final PipeParameters pipeSinkParameters;
   protected final int regionId;
   protected final Supplier<? extends PipeSinkSubtaskExecutor> executor;
 
@@ -43,12 +43,12 @@ public class PipeTaskSinkStage extends PipeTaskStage {
   public PipeTaskSinkStage(
       String pipeName,
       long creationTime,
-      PipeParameters pipeConnectorParameters,
+      PipeParameters pipeSinkParameters,
       int regionId,
       Supplier<? extends PipeSinkSubtaskExecutor> executor) {
     this.pipeName = pipeName;
     this.creationTime = creationTime;
-    this.pipeConnectorParameters = pipeConnectorParameters;
+    this.pipeSinkParameters = pipeSinkParameters;
     this.regionId = regionId;
     this.executor = executor;
 
@@ -60,7 +60,7 @@ protected void registerSubtask() {
         PipeSinkSubtaskManager.instance()
             .register(
                 executor,
-                pipeConnectorParameters,
+                pipeSinkParameters,
                 new PipeTaskSinkRuntimeEnvironment(pipeName, creationTime, regionId));
   }
 
@@ -85,7 +85,7 @@ public void dropSubtask() throws PipeException {
         .deregister(pipeName, creationTime, regionId, connectorSubtaskId);
   }
 
-  public UnboundedBlockingPendingQueue<Event> getPipeConnectorPendingQueue() {
-    return PipeSinkSubtaskManager.instance().getPipeConnectorPendingQueue(connectorSubtaskId);
+  public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue() {
+    return PipeSinkSubtaskManager.instance().getPipeSinkPendingQueue(connectorSubtaskId);
   }
 }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskLifeCycle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskLifeCycle.java
@@ -39,9 +39,9 @@ public class PipeSinkSubtaskLifeCycle implements AutoCloseable {
   protected int registeredTaskCount;
 
   public PipeSinkSubtaskLifeCycle(
-      PipeSinkSubtaskExecutor executor,
-      PipeSinkSubtask subtask,
-      UnboundedBlockingPendingQueue<Event> pendingQueue) {
+      final PipeSinkSubtaskExecutor executor,
+      final PipeSinkSubtask subtask,
+      final UnboundedBlockingPendingQueue<Event> pendingQueue) {
     this.executor = executor;
     this.subtask = subtask;
     this.pendingQueue = pendingQueue;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
@@ -240,7 +240,7 @@ public synchronized void stop(final String attributeSortedString) {
     }
   }
 
-  public UnboundedBlockingPendingQueue<Event> getPipeConnectorPendingQueue(
+  public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(
       final String attributeSortedString) {
     if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
       throw new PipeException(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java
@@ -43,7 +43,7 @@
 public class PipeDataNodeRemainingEventAndTimeOperator extends PipeRemainingOperator {
 
   // Calculate from schema region extractors directly for it requires less computation
-  private final Set<IoTDBSchemaRegionSource> schemaRegionExtractors =
+  private final Set<IoTDBSchemaRegionSource> schemaRegionSources =
       Collections.newSetFromMap(new ConcurrentHashMap<>());
 
   private final AtomicInteger insertNodeEventCount = new AtomicInteger(0);
@@ -105,7 +105,7 @@ public long getRemainingNonHeartbeatEvents() {
         tsfileEventCount.get()
             + rawTabletEventCount.get()
             + insertNodeEventCount.get()
-            + schemaRegionExtractors.stream()
+            + schemaRegionSources.stream()
                 .map(IoTDBSchemaRegionSource::getUnTransferredEventCount)
                 .reduce(Long::sum)
                 .orElse(0L);
@@ -157,7 +157,7 @@ public double getRemainingTime() {
     }
 
     final long totalSchemaRegionWriteEventCount =
-        schemaRegionExtractors.stream()
+        schemaRegionSources.stream()
             .map(IoTDBSchemaRegionSource::getUnTransferredEventCount)
             .reduce(Long::sum)
             .orElse(0L);
@@ -192,8 +192,8 @@ public double getRemainingTime() {
 
   //////////////////////////// Register & deregister (pipe integration) ////////////////////////////
 
-  void register(final IoTDBSchemaRegionSource extractor) {
-    schemaRegionExtractors.add(extractor);
+  void register(final IoTDBSchemaRegionSource source) {
+    schemaRegionSources.add(source);
   }
 
   //////////////////////////// Rate ////////////////////////////
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -323,7 +323,7 @@ private ClusterConfigTaskExecutorHolder() {}
     SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE = SettableFuture.create();
     SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE.setException(
         new IoTDBException(
-            "Subscription not enabled, please set config `subscription_enabled` to true.",
+            "Subscription is not enabled.",
             TSStatusCode.SUBSCRIPTION_NOT_ENABLED_ERROR.getStatusCode()));
   }
 
@@ -1977,15 +1977,15 @@ public SettableFuture<ConfigTaskResult> alterPipe(AlterPipeStatement alterPipeSt
     try {
       if (!alterPipeStatement.getExtractorAttributes().isEmpty()) {
         if (alterPipeStatement.isReplaceAllExtractorAttributes()) {
-          PipeDataNodeAgent.plugin().validateExtractor(alterPipeStatement.getExtractorAttributes());
+          PipeDataNodeAgent.plugin().validateSource(alterPipeStatement.getExtractorAttributes());
         } else {
           pipeMetaFromCoordinator
               .getStaticMeta()
               .getExtractorParameters()
               .addOrReplaceEquivalentAttributes(
                   new PipeParameters(alterPipeStatement.getExtractorAttributes()));
           PipeDataNodeAgent.plugin()
-              .validateExtractor(
+              .validateSource(
                   pipeMetaFromCoordinator.getStaticMeta().getExtractorParameters().getAttribute());
         }
       }
@@ -2008,15 +2008,15 @@ public SettableFuture<ConfigTaskResult> alterPipe(AlterPipeStatement alterPipeSt
       if (!alterPipeStatement.getConnectorAttributes().isEmpty()) {
         if (alterPipeStatement.isReplaceAllConnectorAttributes()) {
           PipeDataNodeAgent.plugin()
-              .validateConnector(pipeName, alterPipeStatement.getConnectorAttributes());
+              .validateSource(pipeName, alterPipeStatement.getConnectorAttributes());
         } else {
           pipeMetaFromCoordinator
               .getStaticMeta()
               .getConnectorParameters()
               .addOrReplaceEquivalentAttributes(
                   new PipeParameters(alterPipeStatement.getConnectorAttributes()));
           PipeDataNodeAgent.plugin()
-              .validateConnector(
+              .validateSource(
                   pipeName,
                   pipeMetaFromCoordinator.getStaticMeta().getConnectorParameters().getAttribute());
         }
@@ -2274,8 +2274,7 @@ public SettableFuture<ConfigTaskResult> createTopic(CreateTopicStatement createT
     final TopicMeta temporaryTopicMeta =
         new TopicMeta(topicName, System.currentTimeMillis(), topicAttributes);
     try {
-      PipeDataNodeAgent.plugin()
-          .validateExtractor(temporaryTopicMeta.generateExtractorAttributes());
+      PipeDataNodeAgent.plugin().validateSource(temporaryTopicMeta.generateExtractorAttributes());
       PipeDataNodeAgent.plugin()
           .validateProcessor(temporaryTopicMeta.generateProcessorAttributes());
     } catch (Exception e) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/stage/SubscriptionTaskSinkStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/stage/SubscriptionTaskSinkStage.java
@@ -31,12 +31,12 @@
 public class SubscriptionTaskSinkStage extends PipeTaskSinkStage {
 
   public SubscriptionTaskSinkStage(
-      String pipeName,
-      long creationTime,
-      PipeParameters pipeConnectorParameters,
-      int regionId,
-      PipeSinkSubtaskExecutor executor) {
-    super(pipeName, creationTime, pipeConnectorParameters, regionId, () -> executor);
+      final String pipeName,
+      final long creationTime,
+      final PipeParameters pipeSinkParameters,
+      final int regionId,
+      final PipeSinkSubtaskExecutor executor) {
+    super(pipeName, creationTime, pipeSinkParameters, regionId, () -> executor);
   }
 
   @Override
@@ -45,7 +45,7 @@ protected void registerSubtask() {
         SubscriptionSinkSubtaskManager.instance()
             .register(
                 executor.get(),
-                pipeConnectorParameters,
+                pipeSinkParameters,
                 new PipeTaskSinkRuntimeEnvironment(pipeName, creationTime, regionId));
   }
 
@@ -70,7 +70,7 @@ public void dropSubtask() throws PipeException {
         .deregister(pipeName, creationTime, regionId, connectorSubtaskId);
   }
 
-  public UnboundedBlockingPendingQueue<Event> getPipeConnectorPendingQueue() {
+  public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue() {
     return SubscriptionSinkSubtaskManager.instance()
         .getPipeConnectorPendingQueue(connectorSubtaskId);
   }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/SubscriptionSinkSubtaskManager.java
@@ -64,10 +64,10 @@ public class SubscriptionSinkSubtaskManager {
 
   public synchronized String register(
       final PipeSinkSubtaskExecutor executor,
-      final PipeParameters pipeConnectorParameters,
+      final PipeParameters pipeSinkParameters,
       final PipeTaskSinkRuntimeEnvironment environment) {
     final String connectorKey =
-        pipeConnectorParameters
+        pipeSinkParameters
             .getStringOrDefault(
                 Arrays.asList(PipeSinkConstant.CONNECTOR_KEY, PipeSinkConstant.SINK_KEY),
                 BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName())
@@ -86,13 +86,13 @@ public synchronized String register(
             environment.getRegionId(),
             connectorKey);
 
-    boolean realTimeFirst =
-        pipeConnectorParameters.getBooleanOrDefault(
+    final boolean realTimeFirst =
+        pipeSinkParameters.getBooleanOrDefault(
             Arrays.asList(
                 PipeSinkConstant.CONNECTOR_REALTIME_FIRST_KEY,
                 PipeSinkConstant.SINK_REALTIME_FIRST_KEY),
             PipeSinkConstant.CONNECTOR_REALTIME_FIRST_DEFAULT_VALUE);
-    String attributeSortedString = generateAttributeSortedString(pipeConnectorParameters);
+    String attributeSortedString = generateAttributeSortedString(pipeSinkParameters);
     attributeSortedString = "__subscription_" + attributeSortedString;
 
     if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
@@ -101,36 +101,33 @@ public synchronized String register(
               ? new PipeRealtimePriorityBlockingQueue()
               : new UnboundedBlockingPendingQueue<>(new PipeDataRegionEventCounter());
 
-      final PipeConnector pipeConnector =
-          PipeDataNodeAgent.plugin().dataRegion().reflectConnector(pipeConnectorParameters);
+      final PipeConnector pipeSink =
+          PipeDataNodeAgent.plugin().dataRegion().reflectSink(pipeSinkParameters);
       // 1. Construct, validate and customize PipeConnector, and then handshake (create connection)
       // with the target
       try {
-        pipeConnector.validate(new PipeParameterValidator(pipeConnectorParameters));
-        pipeConnector.customize(
-            pipeConnectorParameters, new PipeTaskRuntimeConfiguration(environment));
-        pipeConnector.handshake();
+        pipeSink.validate(new PipeParameterValidator(pipeSinkParameters));
+        pipeSink.customize(pipeSinkParameters, new PipeTaskRuntimeConfiguration(environment));
+        pipeSink.handshake();
       } catch (final Exception e) {
         try {
-          pipeConnector.close();
+          pipeSink.close();
         } catch (final Exception closeException) {
           LOGGER.warn(
-              "Failed to close connector after failed to initialize connector. "
-                  + "Ignore this exception.",
+              "Failed to close sink after failed to initialize sink. " + "Ignore this exception.",
               closeException);
         }
-        throw new PipeException(
-            "Failed to construct PipeConnector, because of " + e.getMessage(), e);
+        throw new PipeException("Failed to construct PipeSink, because of " + e.getMessage(), e);
       }
 
       // 2. Fetch topic and consumer group id from connector parameters
-      final String topicName = pipeConnectorParameters.getString(PipeSinkConstant.SINK_TOPIC_KEY);
+      final String topicName = pipeSinkParameters.getString(PipeSinkConstant.SINK_TOPIC_KEY);
       final String consumerGroupId =
-          pipeConnectorParameters.getString(PipeSinkConstant.SINK_CONSUMER_GROUP_KEY);
+          pipeSinkParameters.getString(PipeSinkConstant.SINK_CONSUMER_GROUP_KEY);
       if (Objects.isNull(topicName) || Objects.isNull(consumerGroupId)) {
         throw new SubscriptionException(
             String.format(
-                "Failed to construct subscription connector, because of %s or %s does not exist in pipe connector parameters",
+                "Failed to construct subscription sink, because of %s or %s does not exist in pipe connector parameters",
                 PipeSinkConstant.SINK_TOPIC_KEY, PipeSinkConstant.SINK_CONSUMER_GROUP_KEY));
       }
 
@@ -142,7 +139,7 @@ public synchronized String register(
               attributeSortedString,
               0,
               pendingQueue,
-              pipeConnector,
+              pipeSink,
               topicName,
               consumerGroupId);
       final PipeSinkSubtaskLifeCycle pipeSinkSubtaskLifeCycle =
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/PipePluginAgent.java

Original file line number	Diff line number	Diff line change
`@@ -32,7 +32,7 @@ public PipeConfigNodePluginAgent(PipePluginMetaKeeper pipePluginMetaKeeper) {`
`32`	`32`	`}`
`33`	`33`
`34`	`34`	`@Override`
`35`		`- protected PipeSourceConstructor createPipeExtractorConstructor(`
	`35`	`+ protected PipeSourceConstructor createPipeSourceConstructor(`
`36`	`36`	`PipePluginMetaKeeper pipePluginMetaKeeper) {`
`37`	`37`	`return new PipeConfigRegionSourceConstructor();`
`38`	`38`	`}`
`@@ -44,7 +44,7 @@ protected PipeProcessorConstructor createPipeProcessorConstructor(`
`44`	`44`	`}`
`45`	`45`
`46`	`46`	`@Override`
`47`		`- protected PipeSinkConstructor createPipeConnectorConstructor(`
	`47`	`+ protected PipeSinkConstructor createPipeSinkConstructor(`
`48`	`48`	`PipePluginMetaKeeper pipePluginMetaKeeper) {`
`49`	`49`	`return new PipeConfigRegionSinkConstructor();`
`50`	`50`	`}`
Original file line number	Diff line number	Diff line change
`@@ -183,9 +183,7 @@ private void doTransfer(final PipeConfigRegionWritePlanEvent pipeConfigRegionWri`
`183`	`183`	`true);`
`184`	`184`	`}`
`185`	`185`
`186`		`- if (LOGGER.isDebugEnabled()) {`
`187`		`- LOGGER.debug("Successfully transferred config event {}.", pipeConfigRegionWritePlanEvent);`
`188`		`- }`
	`186`	`+ LOGGER.info("Successfully transferred config event {}.", pipeConfigRegionWritePlanEvent);`
`189`	`187`	`}`
`190`	`188`
`191`	`189`	`private void doTransferWrapper(final PipeConfigRegionSnapshotEvent pipeConfigRegionSnapshotEvent)`
Original file line number	Diff line number	Diff line change
`@@ -212,19 +212,18 @@ public void validate(`
`212`	`212`	`pipeName, extractorAttributes, processorAttributes, connectorAttributes);`
`213`	`213`	`}`
`214`	`214`
`215`		`- public void validateExtractor(Map<String, String> extractorAttributes) throws Exception {`
`216`		`- dataRegionAgent.validateExtractor(extractorAttributes);`
`217`		`- schemaRegionAgent.validateExtractor(extractorAttributes);`
	`215`	`+ public void validateSource(Map<String, String> sourceAttributes) throws Exception {`
	`216`	`+ dataRegionAgent.validateSource(sourceAttributes);`
	`217`	`+ schemaRegionAgent.validateSource(sourceAttributes);`
`218`	`218`	`}`
`219`	`219`
`220`	`220`	`public void validateProcessor(Map<String, String> processorAttributes) throws Exception {`
`221`	`221`	`dataRegionAgent.validateProcessor(processorAttributes);`
`222`	`222`	`schemaRegionAgent.validateProcessor(processorAttributes);`
`223`	`223`	`}`
`224`	`224`
`225`		`- public void validateConnector(String pipeName, Map<String, String> connectorAttributes)`
`226`		`- throws Exception {`
`227`		`- dataRegionAgent.validateConnector(pipeName, connectorAttributes);`
`228`		`- schemaRegionAgent.validateConnector(pipeName, connectorAttributes);`
	`225`	`+ public void validateSource(String pipeName, Map<String, String> sinkAttributes) throws Exception {`
	`226`	`+ dataRegionAgent.validateSink(pipeName, sinkAttributes);`
	`227`	`+ schemaRegionAgent.validateSink(pipeName, sinkAttributes);`
`229`	`228`	`}`
`230`	`229`	`}`
Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@ public PipeDataRegionPluginAgent(DataNodePipePluginMetaKeeper pipePluginMetaKeep`
`33`	`33`	`}`
`34`	`34`
`35`	`35`	`@Override`
`36`		`- protected PipeSourceConstructor createPipeExtractorConstructor(`
	`36`	`+ protected PipeSourceConstructor createPipeSourceConstructor(`
`37`	`37`	`PipePluginMetaKeeper pipePluginMetaKeeper) {`
`38`	`38`	`return new PipeDataRegionSourceConstructor((DataNodePipePluginMetaKeeper) pipePluginMetaKeeper);`
`39`	`39`	`}`
`@@ -46,7 +46,7 @@ protected PipeProcessorConstructor createPipeProcessorConstructor(`
`46`	`46`	`}`
`47`	`47`
`48`	`48`	`@Override`
`49`		`- protected PipeSinkConstructor createPipeConnectorConstructor(`
	`49`	`+ protected PipeSinkConstructor createPipeSinkConstructor(`
`50`	`50`	`PipePluginMetaKeeper pipePluginMetaKeeper) {`
`51`	`51`	`return new PipeDataRegionSinkConstructor((DataNodePipePluginMetaKeeper) pipePluginMetaKeeper);`
`52`	`52`	`}`
Original file line number	Diff line number	Diff line change
`@@ -32,7 +32,7 @@ public PipeSchemaRegionPluginAgent(PipePluginMetaKeeper pipePluginMetaKeeper) {`
`32`	`32`	`}`
`33`	`33`
`34`	`34`	`@Override`
`35`		`- protected PipeSourceConstructor createPipeExtractorConstructor(`
	`35`	`+ protected PipeSourceConstructor createPipeSourceConstructor(`
`36`	`36`	`PipePluginMetaKeeper pipePluginMetaKeeper) {`
`37`	`37`	`return new PipeSchemaRegionSourceConstructor();`
`38`	`38`	`}`
`@@ -44,7 +44,7 @@ protected PipeProcessorConstructor createPipeProcessorConstructor(`
`44`	`44`	`}`
`45`	`45`
`46`	`46`	`@Override`
`47`		`- protected PipeSinkConstructor createPipeConnectorConstructor(`
	`47`	`+ protected PipeSinkConstructor createPipeSinkConstructor(`
`48`	`48`	`PipePluginMetaKeeper pipePluginMetaKeeper) {`
`49`	`49`	`return new PipeSchemaRegionSinkConstructor();`
`50`	`50`	`}`
Original file line number	Diff line number	Diff line change
`@@ -240,7 +240,7 @@ public synchronized void stop(final String attributeSortedString) {`
`240`	`240`	`}`
`241`	`241`	`}`
`242`	`242`
`243`		`- public UnboundedBlockingPendingQueue<Event> getPipeConnectorPendingQueue(`
	`243`	`+ public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(`
`244`	`244`	`final String attributeSortedString) {`
`245`	`245`	`if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {`
`246`	`246`	`throw new PipeException(`