Caideyipi commented on code in PR #18145:
URL: https://github.com/apache/iotdb/pull/18145#discussion_r3549868736
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java:
##########
@@ -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);
Review Comment:
Done in f5d1d33984d. Moved the warning to DataNodePipeMessages (en/zh), and
also localized the new close-timeout warning in this class.
##########
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java:
##########
@@ -68,6 +79,149 @@ public void testDiscardEventsOfPipeDelegatesToConnector() {
}
}
+ @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 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();
+ }
Review Comment:
Done in f5d1d33984d. Replaced the fixed join/sleep check with a completion
latch plus an immediate latch count check before releasing the blocked
handshake. Verified with `mvn -pl iotdb-core/node-commons,iotdb-core/datanode
clean test -Dtest=PipeSinkSubtaskTest -Dsurefire.failIfNoSpecifiedTests=false`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
