Skip to content

Fix pipe drop stuck on blocking sink reconnect#18145

Open
Caideyipi wants to merge 1 commit into
apache:masterfrom
Caideyipi:fix/pipe-drop-stuck-mqtt-sink
Open

Fix pipe drop stuck on blocking sink reconnect#18145
Caideyipi wants to merge 1 commit into
apache:masterfrom
Caideyipi:fix/pipe-drop-stuck-mqtt-sink

Conversation

@Caideyipi

@Caideyipi Caideyipi commented Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Summary

This PR prevents DROP PIPE from getting stuck when an external sink connector blocks during connection retry or close.

Changes:

  • Do not hold the sink subtask monitor while retrying handshake() after PipeConnectionException.
  • Serialize DataNode sink connector operations (transfer, heartbeat, retry handshake, discardEventsOfPipe, and close) with a connector-operation lock, so custom/internal connectors are not required to tolerate concurrent calls from drop/close paths.
  • Keep drop bounded: if a connector operation is already running, connector-side discard/close is deferred instead of blocking the drop thread or calling the connector concurrently.
  • Make sink connector close() bounded by a timeout during drop.
  • Use bounded write-lock acquisition for single pipe meta push and drop handling, consistent with full pipe meta push.
  • Add regression tests for drop/discard not being blocked by connection retry, close not running concurrently with retry handshake, and connector close not waiting forever.

Positive effects

  • DROP PIPE is no longer blocked indefinitely by a sink connector stuck in external network operations such as MQTT/Paho connect or publish retry.
  • DataNode internal RPC/procedure worker threads are less likely to be held forever by pipe cleanup.
  • Drop-time connector cleanup no longer assumes user-defined or internal connectors are thread-safe for concurrent handshake/close/discard calls.
  • Pipe task agent write-lock waits during single meta push/drop become bounded, so ConfigNode can observe a timeout and rely on later metadata synchronization instead of waiting indefinitely.

Potential negative effects

  • DataNode sink connector operations are more strictly serialized. This is safer for connector implementations, but it may slightly delay connector-side discard/close behind a long-running transfer/handshake.
  • If the current connector operation never returns, connector-side discard/close may remain deferred in a daemon close thread; drop still proceeds, but resource release and connector-internal event cleanup can be delayed.
  • If connector close() itself starts and then blocks, drop proceeds after the timeout and the daemon close thread may remain alive until the connector returns.
  • Timed write-lock acquisition may return a pipe meta push/drop timeout under heavy contention; this trades indefinite blocking for fail-fast behavior and relies on later pipe meta synchronization.

Tests

  • mvn spotless:apply -pl iotdb-core/node-commons,iotdb-core/datanode
  • git diff --check
  • $env:MAVEN_OPTS='-Xmx768m'; mvn -pl iotdb-core/node-commons,iotdb-core/datanode -DskipTests "-Dcheckstyle.skip=true" "-Dspotless.check.skip=true" compile
  • $env:MAVEN_OPTS='-Xmx768m'; mvn -pl iotdb-core/node-commons,iotdb-core/datanode -Dtest=PipeSinkSubtaskTest -DfailIfNoTests=false "-Dsurefire.failIfNoSpecifiedTests=false" "-Dcheckstyle.skip=true" "-Dspotless.check.skip=true" test
  • A normal test run without lowering local Maven heap hit Windows native memory/pagefile OOM before executing tests; the targeted test passed with the lower local Maven heap above.

@Caideyipi Caideyipi force-pushed the fix/pipe-drop-stuck-mqtt-sink branch from f7b508e to 63c717d Compare July 8, 2026 03:58
Comment on lines +290 to +294
LOGGER.warn(
"Failed to discard events of pipe {} in connector subtask {}.",
committerKey.getPipeName(),
getDisplayTaskID(),
e);

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

Comment on lines +113 to +128
final Thread discardThread =
new Thread(() -> subtask.discardEventsOfPipe(new CommitterKey("pipe", 1L, 1, -1)));
discardThread.start();
discardThread.join(1000);

try {
Assert.assertFalse(discardThread.isAlive());
Assert.assertFalse(discardEntered.await(100, TimeUnit.MILLISECONDS));
Assert.assertFalse(discardDuringHandshake.get());
} finally {
releaseHandshake.countDown();
failureThread.join(5000);
Assert.assertTrue(discardEntered.await(1, TimeUnit.SECONDS));
Assert.assertFalse(discardDuringHandshake.get());
subtask.close();
}

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.

Fixed-wait may make the test unstable.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants