This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 4482e65468a Fix pipe drop stuck on blocking sink reconnect (#18145)
4482e65468a is described below
commit 4482e65468a0d3699bfed927c8afa293551ae5e2
Author: Caideyipi <[email protected]>
AuthorDate: Fri Jul 10 10:30:00 2026 +0800
Fix pipe drop stuck on blocking sink reconnect (#18145)
* Fix pipe drop stuck on blocking sink reconnect
* Address pipe sink subtask review comments
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 10 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 8 +
.../agent/task/subtask/sink/PipeSinkSubtask.java | 198 ++++++++++++++--
.../task/subtask/sink/PipeSinkSubtaskTest.java | 254 +++++++++++++++++++++
.../commons/pipe/agent/task/PipeTaskAgent.java | 21 +-
.../task/subtask/PipeAbstractSinkSubtask.java | 103 ++++++---
.../pipe/agent/task/subtask/PipeSubtask.java | 2 +-
7 files changed, 533 insertions(+), 63 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index a6d6abbefdb..e7c6c85ca5c 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -228,10 +228,20 @@ public final class DataNodePipeMessages {
"PipeEventCollector: The event {} is already released, skipping it.";
public static final String PIPE_CONNECTOR_SUBTASK_WAS_CLOSED_WITHIN_MS =
"Pipe: connector subtask {} ({}) was closed within {} ms";
+ public static final String
FAILED_TO_DISCARD_EVENTS_OF_PIPE_IN_CONNECTOR_SUBTASK =
+ "Failed to discard events of pipe {} in connector subtask {}.";
public static final String PIPE_META_NOT_FOUND = "Pipe meta not found: ";
public static final String PIPE_SINK_SUBTASKS_WITH_ATTRIBUTES_IS_BOUNDED =
"Pipe sink subtasks with attributes {} is bounded with sinkExecutor {}
and "
+ "callbackExecutor {}.";
+ public static final String PIPE_SINK_SUBTASK_CLOSE_OPERATION_STILL_RUNNING =
+ "is still running";
+ public static final String
+
PIPE_SINK_SUBTASK_CLOSE_OPERATION_WILL_RUN_AFTER_CURRENT_CONNECTOR_OPERATION =
+ "will run after the current connector operation finishes";
+ public static final String PIPE_SINK_SUBTASK_CLOSE_TIMED_OUT =
+ "Timed out after {} ms when closing pipe connector subtask {}. Continue
dropping it. "
+ + "The close operation {}.";
public static final String
PIPE_SINK_SUBTASK_DELAYED_TO_AVOID_FREQUENT_HANDSHAKES =
"Pipe sink subtask {} is delayed for {} ms before polling events to
avoid frequent "
+ "handshakes after client borrow failures.";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index ba2fd769273..54b68d18e0d 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -217,9 +217,17 @@ public final class DataNodePipeMessages {
"PipeEventCollector：事件 {} 已被释放，跳过处理。";
public static final String PIPE_CONNECTOR_SUBTASK_WAS_CLOSED_WITHIN_MS =
"Pipe：connector 子任务 {} ({}) 在 {} ms 内已关闭";
+ public static final String
FAILED_TO_DISCARD_EVENTS_OF_PIPE_IN_CONNECTOR_SUBTASK =
+ "Pipe {} 在 connector 子任务 {} 中的事件丢弃失败。";
public static final String PIPE_META_NOT_FOUND = "未找到 pipe 元数据：";
public static final String PIPE_SINK_SUBTASKS_WITH_ATTRIBUTES_IS_BOUNDED =
"带属性 {} 的 Pipe sink 子任务绑定到 sinkExecutor {} 和 callbackExecutor {}。";
+ public static final String PIPE_SINK_SUBTASK_CLOSE_OPERATION_STILL_RUNNING =
"仍在运行";
+ public static final String
+
PIPE_SINK_SUBTASK_CLOSE_OPERATION_WILL_RUN_AFTER_CURRENT_CONNECTOR_OPERATION =
+ "将在当前 connector 操作完成后运行";
+ public static final String PIPE_SINK_SUBTASK_CLOSE_TIMED_OUT =
+ "关闭 pipe connector 子任务超时 {} ms：{}。继续丢弃该子任务。关闭操作{}。";
public static final String
PIPE_SINK_SUBTASK_DELAYED_TO_AVOID_FREQUENT_HANDSHAKES =
"Pipe sink 子任务 {} 在拉取事件前延迟 {} ms，以避免客户端借用失败后频繁握手。";
public static final String PIPE_SKIPPING_TEMPORARY_TSFILE_WHICH_SHOULDN_T =
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 94e20600909..ce0b2c2233d 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;
@@ -51,6 +52,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 {
@@ -69,6 +75,8 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
// the random delay of the batch transmission. Therefore, here we inject
cron events
// when no event can be pulled.
public static final PipeHeartbeatEvent CRON_HEARTBEAT_EVENT = new
PipeHeartbeatEvent(-1, false);
+ private final ReentrantLock outputPipeSinkOperationLock = new
ReentrantLock();
+ private final Queue<CommitterKey> pendingDiscardCommitterKeys = new
ConcurrentLinkedQueue<>();
public PipeSinkSubtask(
final String taskID,
@@ -132,15 +140,19 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
}
if (event instanceof TabletInsertionEvent) {
- outputPipeSink.transfer((TabletInsertionEvent) event);
- PipeDataRegionSinkMetrics.getInstance().markTabletEvent(taskID);
+ if (executeOutputPipeSinkOperation(
+ () -> outputPipeSink.transfer((TabletInsertionEvent) event))) {
+ PipeDataRegionSinkMetrics.getInstance().markTabletEvent(taskID);
+ }
} else if (event instanceof TsFileInsertionEvent) {
- outputPipeSink.transfer((TsFileInsertionEvent) event);
- PipeDataRegionSinkMetrics.getInstance().markTsFileEvent(taskID);
+ if (executeOutputPipeSinkOperation(
+ () -> outputPipeSink.transfer((TsFileInsertionEvent) event))) {
+ PipeDataRegionSinkMetrics.getInstance().markTsFileEvent(taskID);
+ }
} else if (event instanceof PipeSchemaRegionWritePlanEvent) {
- outputPipeSink.transfer(event);
- if (((PipeSchemaRegionWritePlanEvent) event).getPlanNode().getType()
- != PlanNodeType.DELETE_DATA) {
+ if (executeOutputPipeSinkOperation(() ->
outputPipeSink.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);
@@ -148,10 +160,12 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
} else if (event instanceof PipeHeartbeatEvent) {
transferHeartbeatEvent((PipeHeartbeatEvent) event);
} else {
- outputPipeSink.transfer(
- event instanceof UserDefinedEnrichedEvent
- ? ((UserDefinedEnrichedEvent) event).getUserDefinedEvent()
- : event);
+ executeOutputPipeSinkOperation(
+ () ->
+ outputPipeSink.transfer(
+ event instanceof UserDefinedEnrichedEvent
+ ? ((UserDefinedEnrichedEvent)
event).getUserDefinedEvent()
+ : event));
}
decreaseReferenceCountAndReleaseLastEvent(event, true);
@@ -169,7 +183,15 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
return 0;
}
- return ((PipeSinkWithSchedulingDelay)
outputPipeSink).peekSchedulingDelayMs();
+ outputPipeSinkOperationLock.lock();
+ try {
+ discardPendingEventsOfPipeUnderLock();
+ return isClosed.get()
+ ? 0
+ : ((PipeSinkWithSchedulingDelay)
outputPipeSink).peekSchedulingDelayMs();
+ } finally {
+ outputPipeSinkOperationLock.unlock();
+ }
}
@Override
@@ -178,8 +200,17 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
return 0;
}
- final long remainingSchedulingDelayMs =
- ((PipeSinkWithSchedulingDelay)
outputPipeSink).consumeSchedulingDelayMs();
+ final long remainingSchedulingDelayMs;
+ outputPipeSinkOperationLock.lock();
+ try {
+ discardPendingEventsOfPipeUnderLock();
+ remainingSchedulingDelayMs =
+ isClosed.get()
+ ? 0
+ : ((PipeSinkWithSchedulingDelay)
outputPipeSink).consumeSchedulingDelayMs();
+ } finally {
+ outputPipeSinkOperationLock.unlock();
+ }
if (remainingSchedulingDelayMs <= 0) {
return 0;
}
@@ -201,8 +232,13 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
}
try {
- outputPipeSink.heartbeat();
- outputPipeSink.transfer(event);
+ if (!executeOutputPipeSinkOperation(
+ () -> {
+ outputPipeSink.heartbeat();
+ outputPipeSink.transfer(event);
+ })) {
+ return;
+ }
} catch (final Exception e) {
throw new PipeConnectionException(
String.format(
@@ -218,6 +254,54 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
PipeDataRegionSinkMetrics.getInstance().markPipeHeartbeatEvent(taskID);
}
+ @Override
+ protected boolean handshakeOutputPipeSink() throws Exception {
+ return executeOutputPipeSinkOperation(() -> outputPipeSink.handshake());
+ }
+
+ private boolean executeOutputPipeSinkOperation(final OutputPipeSinkOperation
operation)
+ throws Exception {
+ outputPipeSinkOperationLock.lock();
+ try {
+ discardPendingEventsOfPipeUnderLock();
+ if (isClosed.get()) {
+ return false;
+ }
+
+ operation.execute();
+ discardPendingEventsOfPipeUnderLock();
+ return true;
+ } finally {
+ outputPipeSinkOperationLock.unlock();
+ }
+ }
+
+ private void discardPendingEventsOfPipeUnderLock() {
+ if (!(outputPipeSink instanceof PipeConnectorWithEventDiscard)) {
+ pendingDiscardCommitterKeys.clear();
+ return;
+ }
+
+ CommitterKey committerKey;
+ while ((committerKey = pendingDiscardCommitterKeys.poll()) != null) {
+ try {
+ ((PipeConnectorWithEventDiscard)
outputPipeSink).discardEventsOfPipe(committerKey);
+ } catch (final Exception e) {
+ LOGGER.warn(
+
DataNodePipeMessages.FAILED_TO_DISCARD_EVENTS_OF_PIPE_IN_CONNECTOR_SUBTASK,
+ committerKey.getPipeName(),
+ getDisplayTaskID(),
+ e);
+ }
+ }
+ }
+
+ @FunctionalInterface
+ private interface OutputPipeSinkOperation {
+
+ void execute() throws Exception;
+ }
+
@Override
public void close() {
if (!attributeSortedString.startsWith("schema_")) {
@@ -229,12 +313,13 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
isClosed.set(true);
try {
final long startTime = System.currentTimeMillis();
- outputPipeSink.close();
- LOGGER.info(
- DataNodePipeMessages.PIPE_CONNECTOR_SUBTASK_WAS_CLOSED_WITHIN_MS,
- getDisplayTaskID(),
- outputPipeSink,
- System.currentTimeMillis() - startTime);
+ if (closeOutputPipeSink()) {
+ LOGGER.info(
+ DataNodePipeMessages.PIPE_CONNECTOR_SUBTASK_WAS_CLOSED_WITHIN_MS,
+ getDisplayTaskID(),
+ outputPipeSink,
+ System.currentTimeMillis() - startTime);
+ }
} catch (final Exception e) {
LOGGER.info(
DataNodePipeMessages.EXCEPTION_OCCURRED_WHEN_CLOSING_PIPE_CONNECTOR_SUBTASK,
@@ -249,6 +334,58 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
}
}
+ private boolean closeOutputPipeSink() throws Exception {
+ final AtomicReference<Exception> exception = new AtomicReference<>();
+ final AtomicBoolean closeStarted = new AtomicBoolean(false);
+ final Thread closeThread =
+ new Thread(
+ () -> {
+ outputPipeSinkOperationLock.lock();
+ try {
+ discardPendingEventsOfPipeUnderLock();
+ closeStarted.set(true);
+ outputPipeSink.close();
+ } catch (final Exception e) {
+ exception.set(e);
+ } finally {
+ outputPipeSinkOperationLock.unlock();
+ }
+ },
+ "PipeSinkSubtaskClose-" + getDisplayTaskID());
+ 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(
+ DataNodePipeMessages.PIPE_SINK_SUBTASK_CLOSE_TIMED_OUT,
+ timeoutInMs,
+ getDisplayTaskID(),
+ closeStarted.get()
+ ?
DataNodePipeMessages.PIPE_SINK_SUBTASK_CLOSE_OPERATION_STILL_RUNNING
+ : DataNodePipeMessages
+
.PIPE_SINK_SUBTASK_CLOSE_OPERATION_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.
@@ -298,8 +435,21 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
decreaseHighPriorityTaskCount();
}
- if (outputPipeSink instanceof PipeConnectorWithEventDiscard) {
- ((PipeConnectorWithEventDiscard)
outputPipeSink).discardEventsOfPipe(committerKey);
+ discardOutputPipeSinkEventsOfPipe(committerKey);
+ }
+
+ private void discardOutputPipeSinkEventsOfPipe(final CommitterKey
committerKey) {
+ if (!(outputPipeSink instanceof PipeConnectorWithEventDiscard)) {
+ return;
+ }
+
+ pendingDiscardCommitterKeys.offer(committerKey);
+ if (outputPipeSinkOperationLock.tryLock()) {
+ try {
+ discardPendingEventsOfPipeUnderLock();
+ } finally {
+ outputPipeSinkOperationLock.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 e9d76a4fc7e..c60b9c2c251 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,18 +19,29 @@
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.apache.iotdb.pipe.api.exception.PipeException;
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.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -68,6 +79,156 @@ public class PipeSinkSubtaskTest {
}
}
+ @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",
+ "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",
+ "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",
+ "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);
+ }
+ }
+
@Test
public void testTransferExceptionUsesDisplayTaskID() throws Exception {
final PipeConnector connector = mock(PipeConnector.class);
@@ -104,4 +265,97 @@ public class PipeSinkSubtaskTest {
subtask.close();
}
}
+
+ 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) {
+ this(
+ handshakeEntered,
+ releaseHandshake,
+ new CountDownLatch(0),
+ new AtomicBoolean(false),
+ new CountDownLatch(0),
+ new AtomicBoolean(false));
+ }
+
+ 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 a8763131307..ee465ada69b 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
@@ -165,7 +165,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 {
@@ -368,7 +377,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 9ad6f874eaa..1b51e460956 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
@@ -101,38 +101,24 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
synchronized (this) {
isSubmitted = false;
- if (isClosed.get()) {
- LOGGER.info(PipeMessages.ON_FAILURE_IGNORED_CONNECTOR_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(PipeMessages.ON_FAILURE_IGNORED_EVENT_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(PipeMessages.ON_FAILURE_IGNORED_EVENT_PIPE_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(
@@ -140,19 +126,57 @@ 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,
PipeMessages.NON_CRITICAL_EXCEPTION_WILL_THROW_CRITICAL);
- 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(PipeMessages.ON_FAILURE_IGNORED_CONNECTOR_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(PipeMessages.ON_FAILURE_IGNORED_EVENT_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(PipeMessages.ON_FAILURE_IGNORED_EVENT_PIPE_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,
PipeMessages.NON_CRITICAL_EXCEPTION_WILL_THROW_CRITICAL);
+ super.onFailure(new
PipeRuntimeSinkCriticalException(throwable.getMessage()));
}
}
@@ -169,7 +193,9 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
int retry = 0;
while (retry < MAX_RETRY_TIMES) {
try {
- outputPipeSink.handshake();
+ if (!handshakeOutputPipeSink()) {
+ return false;
+ }
LOGGER.info(PipeMessages.HANDSHAKE_SUCCESS,
outputPipeSink.getClass().getName());
break;
} catch (final Exception e) {
@@ -224,6 +250,11 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
return false;
}
+ protected boolean handshakeOutputPipeSink() throws Exception {
+ outputPipeSink.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 8522fce0cd2..d87fe3a5f38 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
@@ -58,7 +58,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();
