Skip to content

Fix pipe drop stuck on blocking sink reconnect #18145

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Fix pipe drop stuck on blocking sink reconnect #18145

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
197 changes: 173 additions & 24 deletions ...anode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -51,6 +52,11 @@
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 {

Expand All @@ -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,
Expand Down Expand Up @@ -132,26 +140,32 @@ protected boolean executeOnce() {
}

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);
}
} 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);
Expand All @@ -169,7 +183,15 @@ protected long peekSchedulingDelayInMs() {
return 0;
}

return ((PipeSinkWithSchedulingDelay) outputPipeSink).peekSchedulingDelayMs();
outputPipeSinkOperationLock.lock();
try {
discardPendingEventsOfPipeUnderLock();
return isClosed.get()
? 0
: ((PipeSinkWithSchedulingDelay) outputPipeSink).peekSchedulingDelayMs();
} finally {
outputPipeSinkOperationLock.unlock();
}
}

@Override
Expand All @@ -178,8 +200,17 @@ protected long consumeSchedulingDelayInMs() {
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;
}
Expand All @@ -201,8 +232,13 @@ private void transferHeartbeatEvent(final PipeHeartbeatEvent event) {
}

try {
outputPipeSink.heartbeat();
outputPipeSink.transfer(event);
if (!executeOutputPipeSinkOperation(
() -> {
outputPipeSink.heartbeat();
outputPipeSink.transfer(event);
})) {
return;
}
} catch (final Exception e) {
throw new PipeConnectionException(
String.format(
Expand All @@ -218,6 +254,54 @@ private void transferHeartbeatEvent(final PipeHeartbeatEvent event) {
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(
"Failed to discard events of pipe {} in connector subtask {}.",
committerKey.getPipeName(),
getDisplayTaskID(),
e);
Comment on lines +290 to +294

@jt2594838 jt2594838 Jul 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i18n

}
}
}

@FunctionalInterface
private interface OutputPipeSinkOperation {

void execute() throws Exception;
}

@Override
public void close() {
if (!attributeSortedString.startsWith("schema_")) {
Expand All @@ -229,12 +313,13 @@ public void close() {
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,
Expand All @@ -249,6 +334,57 @@ public void close() {
}
}

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(
"Timed out after {} ms when closing pipe connector subtask {}. Continue dropping it. The close operation {}.",
timeoutInMs,
getDisplayTaskID(),
closeStarted.get()
? "is still running"
: "will run after the current connector operation finishes");
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.
Expand Down Expand Up @@ -298,8 +434,21 @@ && isEventFromPipe((EnrichedEvent) lastExceptionEvent, committerKey)) {
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();
}
}
}

Expand Down
Loading
Loading