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 94e206009098..207854cb52ab 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.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 pendingDiscardCommitterKeys = new ConcurrentLinkedQueue<>(); public PipeSinkSubtask( final String taskID, @@ -132,15 +140,19 @@ 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); @@ -148,10 +160,12 @@ protected boolean executeOnce() { } 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 @@ protected long peekSchedulingDelayInMs() { 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 @@ 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; } @@ -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( @@ -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); + } + } + } + + @FunctionalInterface + private interface OutputPipeSinkOperation { + + void execute() throws Exception; + } + @Override public void close() { if (!attributeSortedString.startsWith("schema_")) { @@ -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, @@ -249,6 +334,57 @@ public void close() { } } + private boolean closeOutputPipeSink() throws Exception { + final AtomicReference 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. @@ -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(); + } } } 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 e9d76a4fc7eb..22f59a8bc2b4 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,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(); + } + } + + @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 +258,97 @@ public void testTransferExceptionUsesDisplayTaskID() throws Exception { 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 a87631313074..ee465ada69b8 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 @@ private long convertMsToCeilSeconds(final long timeOutInMs) { 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 @@ private void syncRuntimeExceptionClearTime( 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 9ad6f874eaa3..1b51e460956a 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 void onFailure(final Throwable throwable) { 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 void onFailure(final Throwable throwable) { 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 @@ private boolean onPipeConnectionException(final Throwable throwable) { 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 @@ private boolean onPipeConnectionException(final Throwable throwable) { 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 8522fce0cd24..d87fe3a5f380 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();