This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new d3fb1eac571 Fix pipe drop stuck on blocking sink reconnect (#18145)
(#18188)
d3fb1eac571 is described below
commit d3fb1eac571bfb9e4d6847ac5d849f870a5fc77b
Author: Caideyipi <[email protected]>
AuthorDate: Mon Jul 13 12:10:00 2026 +0800
Fix pipe drop stuck on blocking sink reconnect (#18145) (#18188)
* Fix pipe drop stuck on blocking sink reconnect
* Address pipe sink subtask review comments
(cherry picked from commit 4482e65468a0d3699bfed927c8afa293551ae5e2)
---
.../agent/task/subtask/sink/PipeSinkSubtask.java | 172 +++++++++++++--
.../task/subtask/sink/PipeSinkSubtaskTest.java | 242 +++++++++++++++++++++
.../commons/pipe/agent/task/PipeTaskAgent.java | 21 +-
.../task/subtask/PipeAbstractSinkSubtask.java | 118 ++++++----
.../pipe/agent/task/subtask/PipeSubtask.java | 2 +-
5 files changed, 487 insertions(+), 68 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
index ad8a1273d00..b8c7423c232 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.sink;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
import
org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
@@ -49,6 +50,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Objects;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
@@ -67,6 +73,8 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
// when no event can be pulled.
public static final PipeHeartbeatEvent CRON_HEARTBEAT_EVENT =
new PipeHeartbeatEvent("cron", false);
+ private final ReentrantLock outputPipeConnectorOperationLock = new
ReentrantLock();
+ private final Queue<CommitterKey> pendingDiscardCommitterKeys = new
ConcurrentLinkedQueue<>();
public PipeSinkSubtask(
final String taskID,
@@ -111,15 +119,19 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
}
if (event instanceof TabletInsertionEvent) {
- outputPipeConnector.transfer((TabletInsertionEvent) event);
- PipeDataRegionSinkMetrics.getInstance().markTabletEvent(taskID);
+ if (executeOutputPipeConnectorOperation(
+ () -> outputPipeConnector.transfer((TabletInsertionEvent) event)))
{
+ PipeDataRegionSinkMetrics.getInstance().markTabletEvent(taskID);
+ }
} else if (event instanceof TsFileInsertionEvent) {
- outputPipeConnector.transfer((TsFileInsertionEvent) event);
- PipeDataRegionSinkMetrics.getInstance().markTsFileEvent(taskID);
+ if (executeOutputPipeConnectorOperation(
+ () -> outputPipeConnector.transfer((TsFileInsertionEvent) event)))
{
+ PipeDataRegionSinkMetrics.getInstance().markTsFileEvent(taskID);
+ }
} else if (event instanceof PipeSchemaRegionWritePlanEvent) {
- outputPipeConnector.transfer(event);
- if (((PipeSchemaRegionWritePlanEvent) event).getPlanNode().getType()
- != PlanNodeType.DELETE_DATA) {
+ if (executeOutputPipeConnectorOperation(() ->
outputPipeConnector.transfer(event))
+ && ((PipeSchemaRegionWritePlanEvent) event).getPlanNode().getType()
+ != PlanNodeType.DELETE_DATA) {
// Only plan nodes in schema region will be marked, delete data node
is currently not
// taken into account
PipeSchemaRegionSinkMetrics.getInstance().markSchemaEvent(taskID);
@@ -127,10 +139,12 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
} else if (event instanceof PipeHeartbeatEvent) {
transferHeartbeatEvent((PipeHeartbeatEvent) event);
} else {
- outputPipeConnector.transfer(
- event instanceof UserDefinedEnrichedEvent
- ? ((UserDefinedEnrichedEvent) event).getUserDefinedEvent()
- : event);
+ executeOutputPipeConnectorOperation(
+ () ->
+ outputPipeConnector.transfer(
+ event instanceof UserDefinedEnrichedEvent
+ ? ((UserDefinedEnrichedEvent)
event).getUserDefinedEvent()
+ : event));
}
decreaseReferenceCountAndReleaseLastEvent(event, true);
@@ -149,8 +163,13 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
}
try {
- outputPipeConnector.heartbeat();
- outputPipeConnector.transfer(event);
+ if (!executeOutputPipeConnectorOperation(
+ () -> {
+ outputPipeConnector.heartbeat();
+ outputPipeConnector.transfer(event);
+ })) {
+ return;
+ }
} catch (final Exception e) {
throw new PipeConnectionException(
"PipeConnector: "
@@ -167,6 +186,54 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
PipeDataRegionSinkMetrics.getInstance().markPipeHeartbeatEvent(taskID);
}
+ @Override
+ protected boolean handshakeOutputPipeConnector() throws Exception {
+ return executeOutputPipeConnectorOperation(() ->
outputPipeConnector.handshake());
+ }
+
+ private boolean executeOutputPipeConnectorOperation(final
OutputPipeConnectorOperation operation)
+ throws Exception {
+ outputPipeConnectorOperationLock.lock();
+ try {
+ discardPendingEventsOfPipeUnderLock();
+ if (isClosed.get()) {
+ return false;
+ }
+
+ operation.execute();
+ discardPendingEventsOfPipeUnderLock();
+ return true;
+ } finally {
+ outputPipeConnectorOperationLock.unlock();
+ }
+ }
+
+ private void discardPendingEventsOfPipeUnderLock() {
+ if (!(outputPipeConnector instanceof PipeConnectorWithEventDiscard)) {
+ pendingDiscardCommitterKeys.clear();
+ return;
+ }
+
+ CommitterKey committerKey;
+ while ((committerKey = pendingDiscardCommitterKeys.poll()) != null) {
+ try {
+ ((PipeConnectorWithEventDiscard)
outputPipeConnector).discardEventsOfPipe(committerKey);
+ } catch (final Exception e) {
+ LOGGER.warn(
+ "Failed to discard events of pipe {} in connector subtask {}.",
+ committerKey.getPipeName(),
+ taskID,
+ e);
+ }
+ }
+ }
+
+ @FunctionalInterface
+ private interface OutputPipeConnectorOperation {
+
+ void execute() throws Exception;
+ }
+
@Override
public void close() {
if (!attributeSortedString.startsWith("schema_")) {
@@ -178,12 +245,13 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
isClosed.set(true);
try {
final long startTime = System.currentTimeMillis();
- outputPipeConnector.close();
- LOGGER.info(
- "Pipe: connector subtask {} ({}) was closed within {} ms",
- taskID,
- outputPipeConnector,
- System.currentTimeMillis() - startTime);
+ if (closeOutputPipeConnector()) {
+ LOGGER.info(
+ "Pipe: connector subtask {} ({}) was closed within {} ms",
+ taskID,
+ outputPipeConnector,
+ System.currentTimeMillis() - startTime);
+ }
} catch (final Exception e) {
LOGGER.info(
"Exception occurred when closing pipe connector subtask {}, root
cause: {}",
@@ -198,6 +266,55 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
}
}
+ private boolean closeOutputPipeConnector() throws Exception {
+ final AtomicReference<Exception> exception = new AtomicReference<>();
+ final AtomicBoolean closeStarted = new AtomicBoolean(false);
+ final Thread closeThread =
+ new Thread(
+ () -> {
+ outputPipeConnectorOperationLock.lock();
+ try {
+ discardPendingEventsOfPipeUnderLock();
+ closeStarted.set(true);
+ outputPipeConnector.close();
+ } catch (final Exception e) {
+ exception.set(e);
+ } finally {
+ outputPipeConnectorOperationLock.unlock();
+ }
+ },
+ "PipeSinkSubtaskClose-" + taskID);
+ closeThread.setDaemon(true);
+ closeThread.start();
+
+ final long timeoutInMs =
+ Math.max(
+ 1L,
CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS() * 2L /
3);
+ try {
+ closeThread.join(timeoutInMs);
+ } catch (final InterruptedException e) {
+ closeThread.interrupt();
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+ if (closeThread.isAlive()) {
+ if (closeStarted.get()) {
+ closeThread.interrupt();
+ }
+ LOGGER.warn(
+ "Pipe sink subtask close timed out after {} ms for {}, close
operation {}.",
+ timeoutInMs,
+ taskID,
+ closeStarted.get() ? "is still running" : "will run after current
connector operation");
+ return false;
+ }
+
+ if (exception.get() != null) {
+ throw exception.get();
+ }
+ return true;
+ }
+
/**
* When a pipe is dropped, the connector maybe reused and will not be
closed. So we just discard
* its queued events in the output pipe connector.
@@ -247,8 +364,21 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
decreaseHighPriorityTaskCount();
}
- if (outputPipeConnector instanceof PipeConnectorWithEventDiscard) {
- ((PipeConnectorWithEventDiscard)
outputPipeConnector).discardEventsOfPipe(committerKey);
+ discardOutputPipeConnectorEventsOfPipe(committerKey);
+ }
+
+ private void discardOutputPipeConnectorEventsOfPipe(final CommitterKey
committerKey) {
+ if (!(outputPipeConnector instanceof PipeConnectorWithEventDiscard)) {
+ return;
+ }
+
+ pendingDiscardCommitterKeys.offer(committerKey);
+ if (outputPipeConnectorOperationLock.tryLock()) {
+ try {
+ discardPendingEventsOfPipeUnderLock();
+ } finally {
+ outputPipeConnectorOperationLock.unlock();
+ }
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
index 2a15fb9ea18..7d30078eaa8 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
@@ -19,14 +19,27 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.sink;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import
org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import
org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
import org.apache.iotdb.pipe.api.PipeConnector;
+import
org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
+import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.withSettings;
@@ -60,4 +73,233 @@ public class PipeSinkSubtaskTest {
subtask.close();
}
}
+
+ @Test
+ public void testDiscardEventsOfPipeNotBlockedByConnectionRetry() throws
Exception {
+ final CountDownLatch handshakeEntered = new CountDownLatch(1);
+ final CountDownLatch releaseHandshake = new CountDownLatch(1);
+ final CountDownLatch discardEntered = new CountDownLatch(1);
+ final AtomicBoolean discardDuringHandshake = new AtomicBoolean(false);
+ final PipeConnector connector =
+ new BlockingHandshakeConnector(
+ handshakeEntered,
+ releaseHandshake,
+ new CountDownLatch(0),
+ new AtomicBoolean(false),
+ discardEntered,
+ discardDuringHandshake);
+ final UnboundedBlockingPendingQueue<?> pendingQueue =
mock(UnboundedBlockingPendingQueue.class);
+
+ final PipeSinkSubtask subtask =
+ new PipeSinkSubtask(
+ "PipeSinkSubtaskTest",
+ System.currentTimeMillis(),
+ "data_test",
+ 0,
+ (UnboundedBlockingPendingQueue) pendingQueue,
+ connector);
+
+ final Thread failureThread =
+ new Thread(() -> subtask.onFailure(new
PipeConnectionException("connection broken")));
+ failureThread.start();
+ Assert.assertTrue(handshakeEntered.await(5, TimeUnit.SECONDS));
+
+ final CountDownLatch discardReturned = new CountDownLatch(1);
+ final Thread discardThread =
+ new Thread(
+ () -> {
+ try {
+ subtask.discardEventsOfPipe(new CommitterKey("pipe", 1L, 1,
-1));
+ } finally {
+ discardReturned.countDown();
+ }
+ });
+ discardThread.start();
+
+ try {
+ Assert.assertTrue(discardReturned.await(5, TimeUnit.SECONDS));
+ Assert.assertEquals(1L, discardEntered.getCount());
+ Assert.assertFalse(discardDuringHandshake.get());
+ } finally {
+ releaseHandshake.countDown();
+ failureThread.join(5000);
+ Assert.assertTrue(discardEntered.await(1, TimeUnit.SECONDS));
+ Assert.assertFalse(discardDuringHandshake.get());
+ subtask.close();
+ }
+ }
+
+ @Test
+ public void testCloseNotConcurrentWithConnectionRetry() throws Exception {
+ final int originalTimeout =
+
CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS();
+ CommonDescriptor.getInstance().getConfig().setDnConnectionTimeoutInMS(30);
+
+ final CountDownLatch handshakeEntered = new CountDownLatch(1);
+ final CountDownLatch releaseHandshake = new CountDownLatch(1);
+ final CountDownLatch closeEntered = new CountDownLatch(1);
+ final AtomicBoolean closeDuringHandshake = new AtomicBoolean(false);
+ final PipeConnector connector =
+ new BlockingHandshakeConnector(
+ handshakeEntered,
+ releaseHandshake,
+ closeEntered,
+ closeDuringHandshake,
+ new CountDownLatch(0),
+ new AtomicBoolean(false));
+ final UnboundedBlockingPendingQueue<?> pendingQueue =
mock(UnboundedBlockingPendingQueue.class);
+
+ final PipeSinkSubtask subtask =
+ new PipeSinkSubtask(
+ "PipeSinkSubtaskTest",
+ System.currentTimeMillis(),
+ "data_test",
+ 0,
+ (UnboundedBlockingPendingQueue) pendingQueue,
+ connector);
+
+ final Thread failureThread =
+ new Thread(() -> subtask.onFailure(new
PipeConnectionException("connection broken")));
+ failureThread.start();
+
+ try {
+ Assert.assertTrue(handshakeEntered.await(5, TimeUnit.SECONDS));
+
+ final long startTime = System.currentTimeMillis();
+ subtask.close();
+
+ Assert.assertTrue(System.currentTimeMillis() - startTime < 1000);
+ Assert.assertFalse(closeEntered.await(100, TimeUnit.MILLISECONDS));
+ Assert.assertFalse(closeDuringHandshake.get());
+ } finally {
+ releaseHandshake.countDown();
+ Assert.assertTrue(closeEntered.await(1, TimeUnit.SECONDS));
+ failureThread.join(5000);
+ Assert.assertFalse(closeDuringHandshake.get());
+
CommonDescriptor.getInstance().getConfig().setDnConnectionTimeoutInMS(originalTimeout);
+ }
+ }
+
+ @Test
+ public void testCloseDoesNotWaitForeverForConnectorClose() throws Exception {
+ final int originalTimeout =
+
CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS();
+ CommonDescriptor.getInstance().getConfig().setDnConnectionTimeoutInMS(30);
+
+ final PipeConnector connector = mock(PipeConnector.class);
+ final UnboundedBlockingPendingQueue<?> pendingQueue =
mock(UnboundedBlockingPendingQueue.class);
+ final CountDownLatch closeEntered = new CountDownLatch(1);
+ final CountDownLatch releaseClose = new CountDownLatch(1);
+
+ doAnswer(
+ invocation -> {
+ closeEntered.countDown();
+ releaseClose.await(2, TimeUnit.SECONDS);
+ return null;
+ })
+ .when(connector)
+ .close();
+
+ final PipeSinkSubtask subtask =
+ new PipeSinkSubtask(
+ "PipeSinkSubtaskTest",
+ System.currentTimeMillis(),
+ "data_test",
+ 0,
+ (UnboundedBlockingPendingQueue) pendingQueue,
+ connector);
+
+ try {
+ final long startTime = System.currentTimeMillis();
+ subtask.close();
+
+ Assert.assertTrue(closeEntered.await(1, TimeUnit.SECONDS));
+ Assert.assertTrue(System.currentTimeMillis() - startTime < 1000);
+ } finally {
+ releaseClose.countDown();
+
CommonDescriptor.getInstance().getConfig().setDnConnectionTimeoutInMS(originalTimeout);
+ }
+ }
+
+ private static class BlockingHandshakeConnector
+ implements PipeConnector, PipeConnectorWithEventDiscard {
+
+ private final CountDownLatch handshakeEntered;
+ private final CountDownLatch releaseHandshake;
+ private final CountDownLatch closeEntered;
+ private final AtomicBoolean closeDuringHandshake;
+ private final CountDownLatch discardEntered;
+ private final AtomicBoolean discardDuringHandshake;
+ private volatile boolean handshaking;
+
+ private BlockingHandshakeConnector(
+ final CountDownLatch handshakeEntered,
+ final CountDownLatch releaseHandshake,
+ final CountDownLatch closeEntered,
+ final AtomicBoolean closeDuringHandshake,
+ final CountDownLatch discardEntered,
+ final AtomicBoolean discardDuringHandshake) {
+ this.handshakeEntered = handshakeEntered;
+ this.releaseHandshake = releaseHandshake;
+ this.closeEntered = closeEntered;
+ this.closeDuringHandshake = closeDuringHandshake;
+ this.discardEntered = discardEntered;
+ this.discardDuringHandshake = discardDuringHandshake;
+ }
+
+ @Override
+ public void validate(final PipeParameterValidator validator) throws
Exception {
+ // No-op
+ }
+
+ @Override
+ public void customize(
+ final PipeParameters parameters, final
PipeConnectorRuntimeConfiguration configuration)
+ throws Exception {
+ // No-op
+ }
+
+ @Override
+ public void handshake() throws Exception {
+ handshaking = true;
+ handshakeEntered.countDown();
+ try {
+ releaseHandshake.await(5, TimeUnit.SECONDS);
+ } finally {
+ handshaking = false;
+ }
+ }
+
+ @Override
+ public void heartbeat() throws Exception {
+ // No-op
+ }
+
+ @Override
+ public void transfer(final TabletInsertionEvent tabletInsertionEvent)
throws Exception {
+ // No-op
+ }
+
+ @Override
+ public void transfer(final Event event) throws Exception {
+ // No-op
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (handshaking) {
+ closeDuringHandshake.set(true);
+ }
+ closeEntered.countDown();
+ }
+
+ @Override
+ public void discardEventsOfPipe(
+ final String pipeName, final long creationTime, final int regionId) {
+ if (handshaking) {
+ discardDuringHandshake.set(true);
+ }
+ discardEntered.countDown();
+ }
+ }
}
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 ee629b065df..03906e7e977 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
@@ -163,7 +163,16 @@ public abstract class PipeTaskAgent {
public TPushPipeMetaRespExceptionMessage handleSinglePipeMetaChanges(
final PipeMeta pipeMetaFromCoordinator) {
- acquireWriteLock();
+ final String pipeName =
pipeMetaFromCoordinator.getStaticMeta().getPipeName();
+ if (!tryWriteLockWithTimeOutInMs(
+
CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS() * 2L /
3)) {
+ return new TPushPipeMetaRespExceptionMessage(
+ pipeName,
+ String.format(
+ "Timed out to wait for the pipe task agent write lock when
handling single pipe meta changes for pipe %s.",
+ pipeName),
+ System.currentTimeMillis());
+ }
try {
return handleSinglePipeMetaChangesInternal(pipeMetaFromCoordinator);
} finally {
@@ -355,7 +364,15 @@ public abstract class PipeTaskAgent {
protected abstract void freezeRate(final String pipeName, final long
creationTime);
public TPushPipeMetaRespExceptionMessage handleDropPipe(final String
pipeName) {
- acquireWriteLock();
+ if (!tryWriteLockWithTimeOutInMs(
+
CommonDescriptor.getInstance().getConfig().getDnConnectionTimeoutInMS() * 2L /
3)) {
+ return new TPushPipeMetaRespExceptionMessage(
+ pipeName,
+ String.format(
+ "Timed out to wait for the pipe task agent write lock when
dropping pipe %s.",
+ pipeName),
+ System.currentTimeMillis());
+ }
try {
return handleDropPipeInternal(pipeName);
} finally {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
index 071b5897920..eadd889287c 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
@@ -96,44 +96,24 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
synchronized (this) {
isSubmitted = false;
- if (isClosed.get()) {
- LOGGER.info(
- "onFailure in pipe transfer, ignored because the connector subtask
is dropped.",
- throwable);
- clearReferenceCountAndReleaseLastEvent(null);
- return;
- }
-
- // We assume that the event is cleared as the "lastEvent" in processor
subtask and reaches the
- // connector subtask. Then, it may fail because of released resource and
block the other pipes
- // using the same connector. We simply discard it.
- if (lastExceptionEvent instanceof EnrichedEvent
- && ((EnrichedEvent) lastExceptionEvent).isReleased()) {
- LOGGER.info(
- "onFailure in pipe transfer, ignored because the failure event is
released.",
- throwable);
- submitSelf();
+ if (tryIgnoreFailure(throwable)) {
return;
}
+ }
- // If lastExceptionEvent != lastEvent, it indicates that the lastEvent's
reference has been
- // changed because the pipe of it has been dropped. In that case, we
just discard the event.
- if (lastEvent != lastExceptionEvent) {
- LOGGER.info(
- "onFailure in pipe transfer, ignored because the failure event's
pipe is dropped.",
- throwable);
- clearReferenceCountAndReleaseLastExceptionEvent();
- submitSelf();
+ if (throwable instanceof PipeConnectionException) {
+ // Retry to connect to the target system if the connection is broken. Do
not hold the subtask
+ // lock here because handshaking with an external sink may block for a
long time, while drop
+ // pipe needs the same lock to discard in-flight events.
+ if (onPipeConnectionException(throwable)) {
return;
}
- if (throwable instanceof PipeConnectionException) {
- // Retry to connect to the target system if the connection is broken
- // We should reconstruct the client before re-submit the subtask
- if (onPipeConnectionException(throwable)) {
- // return if the pipe task should be stopped
+ synchronized (this) {
+ if (tryIgnoreFailure(throwable)) {
return;
}
+
if
(PipeConfig.getInstance().isPipeSinkRetryLocallyForConnectionError()) {
super.onFailure(
new PipeRuntimeSinkNonReportTimeConfigurableException(
@@ -141,21 +121,64 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
return;
}
}
+ }
- // Handle exceptions if any available clients exist
- // Notice that the PipeRuntimeConnectorCriticalException must be thrown
here
- // because the upper layer relies on this to stop all the related pipe
tasks
- // Other exceptions may cause the subtask to stop forever and can not be
restarted
- if (throwable instanceof PipeRuntimeSinkCriticalException) {
- super.onFailure(throwable);
- } else {
- // Print stack trace for better debugging
- PipeLogger.log(
- LOGGER::warn,
- throwable,
- "A non PipeRuntimeSinkCriticalException occurred, will throw a
PipeRuntimeSinkCriticalException.");
- super.onFailure(new
PipeRuntimeSinkCriticalException(throwable.getMessage()));
+ synchronized (this) {
+ if (tryIgnoreFailure(throwable)) {
+ return;
}
+ handleFailure(throwable);
+ }
+ }
+
+ private boolean tryIgnoreFailure(final Throwable throwable) {
+ if (isClosed.get()) {
+ LOGGER.info(
+ "onFailure in pipe transfer, ignored because the connector subtask
is dropped.",
+ throwable);
+ clearReferenceCountAndReleaseLastEvent(null);
+ return true;
+ }
+
+ // We assume that the event is cleared as the "lastEvent" in processor
subtask and reaches the
+ // connector subtask. Then, it may fail because of released resource and
block the other pipes
+ // using the same connector. We simply discard it.
+ if (lastExceptionEvent instanceof EnrichedEvent
+ && ((EnrichedEvent) lastExceptionEvent).isReleased()) {
+ LOGGER.info(
+ "onFailure in pipe transfer, ignored because the failure event is
released.", throwable);
+ submitSelf();
+ return true;
+ }
+
+ // If lastExceptionEvent != lastEvent, it indicates that the lastEvent's
reference has been
+ // changed because the pipe of it has been dropped. In that case, we just
discard the event.
+ if (lastEvent != lastExceptionEvent) {
+ LOGGER.info(
+ "onFailure in pipe transfer, ignored because the failure event's
pipe is dropped.",
+ throwable);
+ clearReferenceCountAndReleaseLastExceptionEvent();
+ submitSelf();
+ return true;
+ }
+
+ return false;
+ }
+
+ private void handleFailure(final Throwable throwable) {
+ // Handle exceptions if any available clients exist
+ // Notice that the PipeRuntimeConnectorCriticalException must be thrown
here
+ // because the upper layer relies on this to stop all the related pipe
tasks
+ // Other exceptions may cause the subtask to stop forever and can not be
restarted
+ if (throwable instanceof PipeRuntimeSinkCriticalException) {
+ super.onFailure(throwable);
+ } else {
+ // Print stack trace for better debugging
+ PipeLogger.log(
+ LOGGER::warn,
+ throwable,
+ "A non PipeRuntimeSinkCriticalException occurred, will throw a
PipeRuntimeSinkCriticalException.");
+ super.onFailure(new
PipeRuntimeSinkCriticalException(throwable.getMessage()));
}
}
@@ -171,7 +194,9 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
int retry = 0;
while (retry < MAX_RETRY_TIMES) {
try {
- outputPipeConnector.handshake();
+ if (!handshakeOutputPipeConnector()) {
+ return false;
+ }
LOGGER.info(
"{} handshakes with the target system successfully.",
outputPipeConnector.getClass().getName());
@@ -231,6 +256,11 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
return false;
}
+ protected boolean handshakeOutputPipeConnector() throws Exception {
+ outputPipeConnector.handshake();
+ return true;
+ }
+
private long getHandshakeRetrySleepInterval(final Throwable throwable, final
int retry) {
final long defaultInterval = retry *
PipeConfig.getInstance().getPipeSinkRetryIntervalMs();
return isAuthenticationFailure(throwable)
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
index 2da797c2b3b..1b58d1d6178 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
@@ -55,7 +55,7 @@ public abstract class PipeSubtask
// For fail-over
public static final int MAX_RETRY_TIMES = 5;
protected final AtomicInteger retryCount = new AtomicInteger(0);
- protected Event lastEvent;
+ protected volatile Event lastEvent;
protected PipeSubtask(final String taskID, final long creationTime) {
super();
