(iotdb) branch master updated: Pipe: Added check in PipeSubtask submit logic to ensure one PipeConnectorSubtask instance is running in only one thread (#11474)

[ro...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new 31d0c5bc1e8 Pipe: Added check in PipeSubtask submit logic to ensure one PipeConnectorSubtask instance is running in only one thread (#11474)
31d0c5bc1e8 is described below

commit 31d0c5bc1e84e4782849b015ed3e9f2cf99f2fd0
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Nov 8 21:08:24 2023 +0800

    Pipe: Added check in PipeSubtask submit logic to ensure one PipeConnectorSubtask instance is running in only one thread (#11474)
    
    If the last subtask is running when the subtask is stopped and started again, the subtask will be submitted by the start process and itself when it finishes last round, causing concurrent problems like this:
    
    o.a.t.a.TAsyncClientManager$SelectThread:117 - Ignoring uncaught exception in SelectThread
    java.lang.IndexOutOfBoundsException: Index 1 out of bounds for length 1
    at java.base/jdk.internal.util.Preconditions.outOfBounds(Preconditions.java:64)
    at java.base/jdk.internal.util.Preconditions.outOfBoundsCheckIndex(Preconditions.java:70)
    at java.base/jdk.internal.util.Preconditions.checkIndex(Preconditions.java:248)
    at java.base/java.util.Objects.checkIndex(Objects.java:372)
    at java.base/java.util.ArrayList.get(ArrayList.java:459)
    at org.apache.iotdb.db.pipe.connector.protocol.thrift.async.handler.PipeTransferTabletBatchEventHandler.onError(PipeTransferTabletBatchEventHandler.java:95)
    at org.apache.thrift.async.TAsyncMethodCall.onError(TAsyncMethodCall.java:216)
    at org.apache.thrift.async.TAsyncMethodCall.transition(TAsyncMethodCall.java:210)
    at org.apache.thrift.async.TAsyncClientManager$SelectThread.transitionMethods(TAsyncClientManager.java:143)
    at org.apache.thrift.async.TAsyncClientManager$SelectThread.run(TAsyncClientManager.java:113)
    
    This is because two threads share the same batch builder in async connector, and one is using the copy of its elements to create handler while one is adding elements to the batch, resulting to elements number inequality in handler due to some unadded elements in batch builder.
    
    This commit solved the problem and allow only 1 thread to pick the subtask.
    
    ---------
    
    Co-authored-by: Steve Yurong Su <ro...@apache.org>
---
 .../iotdb/db/pipe/task/subtask/PipeSubtask.java    |  4 ++++
 .../subtask/connector/PipeConnectorSubtask.java    | 28 ++++++++++++++++++----
 2 files changed, 27 insertions(+), 5 deletions(-)

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
index 4614a25d8c5..352dbf091b7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/PipeSubtask.java
@@ -198,6 +198,10 @@ public abstract class PipeSubtask
     }
   }
 
+  /**
+   * Submit the subTask. Be sure to add parallel check since a subtask is currently not designed to
+   * run in parallel.
+   */
   public abstract void submitSelf();
 
   public void allowSubmittingSelf() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java
index 283a3f71510..e2fafe00daf 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/connector/PipeConnectorSubtask.java
@@ -59,6 +59,10 @@ public class PipeConnectorSubtask extends PipeSubtask {
   protected final DecoratingLock callbackDecoratingLock = new DecoratingLock();
   protected ExecutorService subtaskCallbackListeningExecutor;
 
+  // For controlling subtask submitting, making sure that a subtask is submitted to only one thread
+  // at a time
+  protected volatile boolean isSubmitted = false;
+
   // Record these variables to provide corresponding value to tag key of monitoring metrics
   private final String attributeSortedString;
   private final int connectorIndex;
@@ -159,7 +163,16 @@ public class PipeConnectorSubtask extends PipeSubtask {
   }
 
   @Override
-  public void onFailure(@NotNull Throwable throwable) {
+  public synchronized void onSuccess(Boolean hasAtLeastOneEventProcessed) {
+    isSubmitted = false;
+
+    super.onSuccess(hasAtLeastOneEventProcessed);
+  }
+
+  @Override
+  public synchronized void onFailure(@NotNull Throwable throwable) {
+    isSubmitted = false;
+
     if (isClosed.get()) {
       LOGGER.info("onFailure in pipe transfer, ignored because pipe is dropped.");
       releaseLastEvent(false);
@@ -227,8 +240,6 @@ public class PipeConnectorSubtask extends PipeSubtask {
               MAX_RETRY_TIMES,
               taskID,
               throwable);
-
-          // FIXME: non-EnrichedEvent should be reported to the ConfigNode instead of being logged
         }
 
         // Although the pipe task will be stopped, we still don't release the last event here
@@ -250,9 +261,15 @@ public class PipeConnectorSubtask extends PipeSubtask {
     super.onFailure(new PipeRuntimeConnectorCriticalException(throwable.getMessage()));
   }
 
+  /**
+   * Submit a subTask to the executor to keep it running. Note that the function will be called when
+   * connector starts or the subTask finishes the last round, Thus the "isRunning" sign is added to
+   * avoid concurrent problem of the two, ensuring two or more submitting threads generates only one
+   * winner.
+   */
   @Override
-  public void submitSelf() {
-    if (shouldStopSubmittingSelf.get()) {
+  public synchronized void submitSelf() {
+    if (shouldStopSubmittingSelf.get() || isSubmitted) {
       return;
     }
 
@@ -260,6 +277,7 @@ public class PipeConnectorSubtask extends PipeSubtask {
     try {
       final ListenableFuture<Boolean> nextFuture = subtaskWorkerThreadPoolExecutor.submit(this);
       Futures.addCallback(nextFuture, this, subtaskCallbackListeningExecutor);
+      isSubmitted = true;
     } finally {
       callbackDecoratingLock.markAsDecorated();
     }