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 <[email protected]>
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 <[email protected]>
---
.../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();
}
