You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/01/15 11:29:23 UTC
(iotdb) branch master updated: Pipe: Fix deadlock when PipeTaskAgent.handlePipeMetaChanges and PipeTaskDataNodeAgent.stopAllPipesWithCriticalException are invoked concurrently & Fix concurrent issues caused by addFailureEventToRetryQueue & transferQueuedEventsIfNecessary (#11893)
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 bcfeafe1838 Pipe: Fix deadlock when PipeTaskAgent.handlePipeMetaChanges and PipeTaskDataNodeAgent.stopAllPipesWithCriticalException are invoked concurrently & Fix concurrent issues caused by addFailureEventToRetryQueue & transferQueuedEventsIfNecessary (#11893)
bcfeafe1838 is described below
commit bcfeafe18383de4779057a29bdc78e4cd1b7c791
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Jan 15 19:29:17 2024 +0800
Pipe: Fix deadlock when PipeTaskAgent.handlePipeMetaChanges and PipeTaskDataNodeAgent.stopAllPipesWithCriticalException are invoked concurrently & Fix concurrent issues caused by addFailureEventToRetryQueue & transferQueuedEventsIfNecessary (#11893)
1. Fix deadlock when PipeTaskAgent.handlePipeMetaChanges and PipeTaskDataNodeAgent.stopAllPipesWithCriticalException are invoked concurrently
```
Found one Java-level deadlock:
=============================
"pool-41-IoTDB-DataNodeInternalRPC-Processor-2":
waiting to lock monitor 0x00000299461ba110 (object 0x000000054e91ea58, a org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask),
which is held by "pool-25-IoTDB-Pipe-SubTask-Callback-Executor-Pool-1"
"pool-25-IoTDB-Pipe-SubTask-Callback-Executor-Pool-1":
waiting to lock monitor 0x000002994731e980 (object 0x0000000404acc078, a org.apache.iotdb.db.pipe.agent.task.PipeTaskDataNodeAgent),
which is held by "pool-41-IoTDB-DataNodeInternalRPC-Processor-2"
Java stack information for the threads listed above:
===================================================
"pool-41-IoTDB-DataNodeInternalRPC-Processor-2":
at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.submitSelf(PipeConnectorSubtask.java:297)
- waiting to lock <0x000000054e91ea58> (a org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask)
at org.apache.iotdb.commons.pipe.execution.executor.PipeSubtaskExecutor.start(PipeSubtaskExecutor.java:90)
- locked <0x000000054e91e840> (a org.apache.iotdb.db.pipe.execution.executor.dataregion.PipeDataRegionConnectorSubtaskExecutor)
at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtaskLifeCycle.start(PipeConnectorSubtaskLifeCycle.java:121)
- locked <0x000000054e02eb90> (a org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtaskLifeCycle)
at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtaskManager.start(PipeConnectorSubtaskManager.java:155)
- locked <0x000000054e02ea40> (a org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtaskManager)
at org.apache.iotdb.db.pipe.task.stage.PipeTaskConnectorStage.startSubtask(PipeTaskConnectorStage.java:66)
at org.apache.iotdb.commons.pipe.task.stage.PipeTaskStage.start(PipeTaskStage.java:86)
- locked <0x000000054ee138f8> (a org.apache.iotdb.db.pipe.task.stage.PipeTaskConnectorStage)
at org.apache.iotdb.db.pipe.task.PipeDataNodeTask.start(PipeDataNodeTask.java:67)
at org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent.startPipe(PipeTaskAgent.java:513)
at org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent.executeSinglePipeRuntimeMetaChanges(PipeTaskAgent.java:248)
at org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent.executeSinglePipeMetaChanges(PipeTaskAgent.java:168)
at org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent.handlePipeMetaChangesInternal(PipeTaskAgent.java:325)
at org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent.handlePipeMetaChanges(PipeTaskAgent.java:306)
- locked <0x0000000404acc078> (a org.apache.iotdb.db.pipe.agent.task.PipeTaskDataNodeAgent)
at org.apache.iotdb.db.protocol.thrift.impl.DataNodeInternalRPCServiceImpl.pushPipeMeta(DataNodeInternalRPCServiceImpl.java:961)
at org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService$Processor$pushPipeMeta.getResult(IDataNodeRPCService.java:5746)
at org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService$Processor$pushPipeMeta.getResult(IDataNodeRPCService.java:5726)
at org.apache.thrift.ProcessFunction.process(ProcessFunction.java:38)
at org.apache.thrift.TBaseProcessor.process(TBaseProcessor.java:38)
at org.apache.thrift.server.TThreadPoolServer$WorkerProcess.run(TThreadPoolServer.java:248)
at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@17.0.8/ThreadPoolExecutor.java:1136)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@17.0.8/ThreadPoolExecutor.java:635)
at java.lang.Thread.run(java.base@17.0.8/Thread.java:833)
"pool-25-IoTDB-Pipe-SubTask-Callback-Executor-Pool-1":
at org.apache.iotdb.db.pipe.agent.task.PipeTaskDataNodeAgent.stopAllPipesWithCriticalException(PipeTaskDataNodeAgent.java:79)
- waiting to lock <0x0000000404acc078> (a org.apache.iotdb.db.pipe.agent.task.PipeTaskDataNodeAgent)
at org.apache.iotdb.db.pipe.agent.runtime.PipeRuntimeAgent.report(PipeRuntimeAgent.java:142)
at org.apache.iotdb.db.pipe.event.EnrichedEvent.reportException(EnrichedEvent.java:220)
at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.onFailure(PipeConnectorSubtask.java:253)
- locked <0x000000054e91ea58> (a org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask)
at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1119)
at org.apache.iotdb.commons.concurrent.WrappedRunnable$1.runMayThrow(WrappedRunnable.java:45)
at org.apache.iotdb.commons.concurrent.WrappedRunnable.run(WrappedRunnable.java:30)
at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@17.0.8/ThreadPoolExecutor.java:1136)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@17.0.8/ThreadPoolExecutor.java:635)
at java.lang.Thread.run(java.base@17.0.8/Thread.java:833)
Found 1 deadlock.
```
2. Fix concurrent issues caused by addFailureEventToRetryQueue & transferQueuedEventsIfNecessary
```
2024-01-15 15:13:53,529 [pool-50-IoTDB-Pipe-DataRegion-Connector-Executor-Pool-4] INFO o.a.i.d.p.c.p.t.s.IoTDBThriftSyncConnector:471 - Successfully transferred file /data/iotdb/apache-iotdb-1.3.1-SNAPSHOT-all-bin/data/datanode/data/pipe/tsfile/sequence-root.test.g_0-2-2721-1705043651661-61-1-0.tsfile.
...
2024-01-15 15:14:01,200 [pool-2-IoTDB-Pipe-Runtime-Periodical-Job-Executor-1] INFO o.a.i.d.p.r.t.PipeTsFileResource:126 - PipeTsFileResource: Closed tsfile /data/iotdb/apache-iotdb-1.3.1-SNAPSHOT-all-bin/data/datanode/data/pipe/tsfile/sequence-root.test.g_0-2-2721-1705043651661-61-1-0.tsfile and cleaned up.
...
2024-01-15 15:14:05,702 [pool-50-IoTDB-Pipe-DataRegion-Connector-Executor-Pool-4] ERROR o.a.i.c.c.t.WrappedThreadPoolExecutor:129 - Exception in thread pool org.apache.iotdb.threadpool:type=Pipe-DataRegion-Connector-Executor-Pool
org.apache.iotdb.pipe.api.exception.PipeConnectionException: Failed to transfer tsfile insertion event PipeTsFileInsertionEvent{resource=file is /data/iotdb/apache-iotdb-1.3.1-SNAPSHOT-all-bin/data/datanode/data/sequence/root.test.g_0/2/2721/1705043651661-61-1-0.tsfile, status: NORMAL, tsFile=/data/iotdb/apache-iotdb-1.3.1-SNAPSHOT-all-bin/data/datanode/data/pipe/tsfile/sequence-root.test.g_0-2-2721-1705043651661-61-1-0.tsfile, isClosed=true}, because Seal file /data/iotdb/apache-iotd [...]
at org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBThriftSyncConnector.transfer(IoTDBThriftSyncConnector.java:272)
at org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBThriftAsyncConnector.transferQueuedEventsIfNecessary(IoTDBThriftAsyncConnector.java:351)
at org.apache.iotdb.db.pipe.connector.protocol.thrift.async.IoTDBThriftAsyncConnector.transfer(IoTDBThriftAsyncConnector.java:250)
at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.executeOnce(PipeConnectorSubtask.java:140)
at org.apache.iotdb.commons.pipe.task.subtask.PipeSubtask.call(PipeSubtask.java:75)
at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.call(PipeConnectorSubtask.java:108)
at org.apache.iotdb.db.pipe.task.subtask.connector.PipeConnectorSubtask.call(PipeConnectorSubtask.java:51)
at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131)
at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:75)
at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.iotdb.pipe.api.exception.PipeException: Seal file /data/iotdb/apache-iotdb-1.3.1-SNAPSHOT-all-bin/data/datanode/data/pipe/tsfile/sequence-root.test.g_0-2-2721-1705043651661-61-1-0.tsfile error, result status TSStatus(code:1804, message:Failed to seal file sequence-root.test.g_0-2-2721-1705043651661-61-1-0.tsfile, because the length of file is not correct. The original file has length 0, but receiver file has length 117603772.).
at org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBThriftSyncConnector.doTransfer(IoTDBThriftSyncConnector.java:469)
at org.apache.iotdb.db.pipe.connector.protocol.thrift.sync.IoTDBThriftSyncConnector.transfer(IoTDBThriftSyncConnector.java:269)
... 12 common frames omitted
```
---
.../pipe/coordinator/runtime/PipeMetaSyncer.java | 3 ++
.../db/pipe/agent/runtime/PipeRuntimeAgent.java | 4 ++-
.../db/pipe/agent/task/PipeTaskDataNodeAgent.java | 41 ++++++++++++++++++----
.../thrift/async/IoTDBThriftAsyncConnector.java | 38 +++++++++++++-------
.../commons/pipe/agent/task/PipeTaskAgent.java | 20 ++++++++---
.../commons/pipe/task/meta/PipeMetaKeeper.java | 4 +++
.../commons/pipe/task/subtask/PipeSubtask.java | 15 +++++++-
7 files changed, 98 insertions(+), 27 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java
index 6d3f0674817..c80b1018ca3 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java
@@ -114,6 +114,9 @@ public class PipeMetaSyncer {
&& pipeAutoRestartRoundCounter.incrementAndGet()
== PipeConfig.getInstance().getPipeMetaSyncerAutoRestartPipeCheckIntervalRound()) {
somePipesNeedRestarting = autoRestartWithLock();
+ if (somePipesNeedRestarting) {
+ LOGGER.info("Some pipes need restarting, will restart them after this sync");
+ }
pipeAutoRestartRoundCounter.set(0);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
index 991c57b3a15..2414613ad96 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
@@ -39,6 +39,7 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
public class PipeRuntimeAgent implements IService {
@@ -139,7 +140,8 @@ public class PipeRuntimeAgent implements IService {
// Quick stop all pipes locally if critical exception occurs,
// no need to wait for the next heartbeat cycle.
if (pipeRuntimeException instanceof PipeRuntimeCriticalException) {
- PipeAgent.task().stopAllPipesWithCriticalException();
+ // To avoid deadlock, we use a new thread to stop all pipes.
+ CompletableFuture.runAsync(() -> PipeAgent.task().stopAllPipesWithCriticalException());
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskDataNodeAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskDataNodeAgent.java
index 4ac18a1168e..15741629389 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskDataNodeAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskDataNodeAgent.java
@@ -75,12 +75,39 @@ public class PipeTaskDataNodeAgent extends PipeTaskAgent {
return new PipeDataNodeBuilder(pipeMetaFromConfigNode).build();
}
- public synchronized void stopAllPipesWithCriticalException() {
- acquireWriteLock();
+ /**
+ * Using try lock method to prevent deadlock when stopping all pipes with critical exceptions and
+ * {@link PipeTaskDataNodeAgent#handlePipeMetaChanges(List)}} concurrently.
+ */
+ public void stopAllPipesWithCriticalException() {
try {
- stopAllPipesWithCriticalExceptionInternal();
- } finally {
- releaseWriteLock();
+ int retryCount = 0;
+ while (true) {
+ if (tryWriteLockWithTimeOut(5)) {
+ try {
+ stopAllPipesWithCriticalExceptionInternal();
+ LOGGER.info("Stopped all pipes with critical exception.");
+ return;
+ } finally {
+ releaseWriteLock();
+ }
+ } else {
+ Thread.sleep(1000);
+ LOGGER.warn(
+ "Failed to stop all pipes with critical exception, retry count: {}.", ++retryCount);
+ }
+ }
+ } catch (InterruptedException e) {
+ LOGGER.error(
+ "Interrupted when trying to stop all pipes with critical exception, exception message: {}",
+ e.getMessage(),
+ e);
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ LOGGER.error(
+ "Failed to stop all pipes with critical exception, exception message: {}",
+ e.getMessage(),
+ e);
}
}
@@ -220,7 +247,7 @@ public class PipeTaskDataNodeAgent extends PipeTaskAgent {
///////////////////////// Heartbeat /////////////////////////
- public synchronized void collectPipeMetaList(TDataNodeHeartbeatResp resp) throws TException {
+ public void collectPipeMetaList(TDataNodeHeartbeatResp resp) throws TException {
// Try the lock instead of directly acquire it to prevent the block of the cluster heartbeat
// 10s is the half of the HEARTBEAT_TIMEOUT_TIME defined in class BaseNodeCache in ConfigNode
if (!tryReadLockWithTimeOut(10)) {
@@ -251,7 +278,7 @@ public class PipeTaskDataNodeAgent extends PipeTaskAgent {
resp.setPipeMetaList(pipeMetaBinaryList);
}
- public synchronized void collectPipeMetaList(TPipeHeartbeatReq req, TPipeHeartbeatResp resp)
+ public void collectPipeMetaList(TPipeHeartbeatReq req, TPipeHeartbeatResp resp)
throws TException {
acquireReadLock();
try {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
index bed003ec792..185699cf3d9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
@@ -341,25 +341,35 @@ public class IoTDBThriftAsyncConnector extends IoTDBConnector {
*/
private synchronized void transferQueuedEventsIfNecessary() throws Exception {
while (!retryEventQueue.isEmpty()) {
- final Event event = retryEventQueue.peek();
-
- if (event instanceof PipeInsertNodeTabletInsertionEvent) {
- retryConnector.transfer((PipeInsertNodeTabletInsertionEvent) event);
- } else if (event instanceof PipeRawTabletInsertionEvent) {
- retryConnector.transfer((PipeRawTabletInsertionEvent) event);
- } else if (event instanceof PipeTsFileInsertionEvent) {
- retryConnector.transfer((PipeTsFileInsertionEvent) event);
+ final Event peekedEvent = retryEventQueue.peek();
+
+ if (peekedEvent instanceof PipeInsertNodeTabletInsertionEvent) {
+ retryConnector.transfer((PipeInsertNodeTabletInsertionEvent) peekedEvent);
+ } else if (peekedEvent instanceof PipeRawTabletInsertionEvent) {
+ retryConnector.transfer((PipeRawTabletInsertionEvent) peekedEvent);
+ } else if (peekedEvent instanceof PipeTsFileInsertionEvent) {
+ retryConnector.transfer((PipeTsFileInsertionEvent) peekedEvent);
} else {
LOGGER.warn(
- "IoTDBThriftAsyncConnector does not support transfer generic event: {}.", event);
+ "IoTDBThriftAsyncConnector does not support transfer generic event: {}.", peekedEvent);
}
- if (event instanceof EnrichedEvent) {
- ((EnrichedEvent) event)
+ if (peekedEvent instanceof EnrichedEvent) {
+ ((EnrichedEvent) peekedEvent)
.decreaseReferenceCount(IoTDBThriftAsyncConnector.class.getName(), true);
}
- retryEventQueue.poll();
+ final Event polledEvent = retryEventQueue.poll();
+ if (polledEvent != peekedEvent) {
+ LOGGER.error(
+ "The event polled from the queue is not the same as the event peeked from the queue. "
+ + "Peeked event: {}, polled event: {}.",
+ peekedEvent,
+ polledEvent);
+ }
+ if (polledEvent != null) {
+ LOGGER.info("Polled event {} from retry queue.", polledEvent);
+ }
}
}
@@ -379,8 +389,10 @@ public class IoTDBThriftAsyncConnector extends IoTDBConnector {
*
* @param event event to retry
*/
- public void addFailureEventToRetryQueue(Event event) {
+ public synchronized void addFailureEventToRetryQueue(Event event) {
retryEventQueue.offer(event);
+
+ LOGGER.info("Added event {} to retry queue.", event);
}
//////////////////////////// Operations for close ////////////////////////////
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
index 79575a6bb17..0a32cd85dfc 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
@@ -81,7 +81,7 @@ public abstract class PipeTaskAgent {
return pipeMetaKeeper.tryReadLock(timeOutInSeconds);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- LOGGER.warn("Interruption during requiring pipeMetaKeeper lock.", e);
+ LOGGER.warn("Interruption during requiring pipeMetaKeeper read lock.", e);
return false;
}
}
@@ -94,13 +94,23 @@ public abstract class PipeTaskAgent {
pipeMetaKeeper.acquireWriteLock();
}
+ protected boolean tryWriteLockWithTimeOut(long timeOutInSeconds) {
+ try {
+ return pipeMetaKeeper.tryWriteLock(timeOutInSeconds);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn("Interruption during requiring pipeMetaKeeper write lock.", e);
+ return false;
+ }
+ }
+
protected void releaseWriteLock() {
pipeMetaKeeper.releaseWriteLock();
}
////////////////////////// Pipe Task Management Entry //////////////////////////
- public synchronized TPushPipeMetaRespExceptionMessage handleSinglePipeMetaChanges(
+ public TPushPipeMetaRespExceptionMessage handleSinglePipeMetaChanges(
PipeMeta pipeMetaFromCoordinator) {
acquireWriteLock();
try {
@@ -272,7 +282,7 @@ public abstract class PipeTaskAgent {
}
}
- public synchronized TPushPipeMetaRespExceptionMessage handleDropPipe(String pipeName) {
+ public TPushPipeMetaRespExceptionMessage handleDropPipe(String pipeName) {
acquireWriteLock();
try {
return handleDropPipeInternal(pipeName);
@@ -299,7 +309,7 @@ public abstract class PipeTaskAgent {
}
}
- public synchronized List<TPushPipeMetaRespExceptionMessage> handlePipeMetaChanges(
+ public List<TPushPipeMetaRespExceptionMessage> handlePipeMetaChanges(
List<PipeMeta> pipeMetaListFromCoordinator) {
acquireWriteLock();
try {
@@ -356,7 +366,7 @@ public abstract class PipeTaskAgent {
return exceptionMessages;
}
- public synchronized void dropAllPipeTasks() {
+ public void dropAllPipeTasks() {
acquireWriteLock();
try {
dropAllPipeTasksInternal();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaKeeper.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaKeeper.java
index 86b930b3b04..003eec9bdd8 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaKeeper.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaKeeper.java
@@ -59,6 +59,10 @@ public class PipeMetaKeeper {
pipeMetaKeeperLock.writeLock().lock();
}
+ public boolean tryWriteLock(long timeOut) throws InterruptedException {
+ return pipeMetaKeeperLock.writeLock().tryLock(timeOut, TimeUnit.SECONDS);
+ }
+
public void releaseWriteLock() {
pipeMetaKeeperLock.writeLock().unlock();
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/subtask/PipeSubtask.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/subtask/PipeSubtask.java
index c8e996478ee..98fd635b2be 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/subtask/PipeSubtask.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/task/subtask/PipeSubtask.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.pipe.api.event.Event;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListeningExecutorService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
@@ -33,6 +35,8 @@ import java.util.concurrent.atomic.AtomicInteger;
public abstract class PipeSubtask
implements FutureCallback<Boolean>, Callable<Boolean>, AutoCloseable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(PipeSubtask.class);
+
// Used for identifying the subtask
protected final String taskID;
@@ -101,8 +105,17 @@ public abstract class PipeSubtask
@Override
public synchronized void onSuccess(Boolean hasAtLeastOneEventProcessed) {
- retryCount.set(0);
+ final int totalRetryCount = retryCount.getAndSet(0);
+
submitSelf();
+
+ if (totalRetryCount != 0) {
+ LOGGER.warn(
+ "Successfully executed subtask {}({}) after {} retries.",
+ taskID,
+ this.getClass().getSimpleName(),
+ totalRetryCount);
+ }
}
/**