From 24fbeba2e8e3c4c7884e4800c4c699a0e0533402 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 16 Jul 2026 11:59:08 +0800 Subject: [PATCH] Fix unbounded async pipe sink retries --- .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../task/subtask/sink/PipeSinkSubtask.java | 3 + .../async/IoTDBDataRegionAsyncSink.java | 178 +++++++++++++++--- .../handler/PipeTransferTrackableHandler.java | 30 ++- .../subtask/sink/PipeSinkSubtaskTest.java | 44 +++++ .../PipeTransferTrackableHandlerTest.java | 76 ++++++++ .../iotdb/commons/i18n/ConfigMessages.java | 4 + .../iotdb/commons/i18n/PipeMessages.java | 4 + .../iotdb/commons/i18n/ConfigMessages.java | 4 + .../iotdb/commons/i18n/PipeMessages.java | 4 + .../iotdb/commons/conf/CommonConfig.java | 30 +++ .../iotdb/commons/pipe/config/PipeConfig.java | 14 ++ .../commons/pipe/config/PipeDescriptor.java | 14 ++ 14 files changed, 381 insertions(+), 28 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 af39e7b4b4208..bfd57672ee3d4 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 @@ -2543,6 +2543,8 @@ private DataNodePipeMessages() {} "Incomplete timestamps in current tablet format deserialization."; public static final String MESSAGE_RECEIVER_ARG_IS_TEMPORARILY_UNAVAILABLE_THROTTLE_REQUESTS_FOR_ARG_MS_STATUS_ARG_F37192D9 = "Receiver {} is temporarily unavailable, throttle requests for {} ms. Status: {}"; + public static final String EXCEPTION_RECEIVER_ARG_REMAINED_TEMPORARILY_UNAVAILABLE_FOR_MORE_THAN_ARG_MS_PAUSE_REGULAR_RETRIES_AND_PROBE_EVERY_ARG_MS_C515DD97 = + "Receiver %s remained temporarily unavailable for more than %d ms, pause regular retries and probe every %d ms."; public static final String MESSAGE_SUCCESSFULLY_TRANSFERRED_BATCHED_SCHEMA_EVENTS_BATCH_SIZE_ARG_CF2E881C = "Successfully transferred batched schema events, batch size {}."; public static final String EXCEPTION_AUTO_CREATE_TREE_DATABASE_FAILED_ARG_STATUS_CODE_ARG_C6175C27 = 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 138f0c752db97..2b97336eaba9e 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 @@ -2372,6 +2372,8 @@ private DataNodePipeMessages() {} "当前 tablet 格式反序列化中时间戳不完整。"; public static final String MESSAGE_RECEIVER_ARG_IS_TEMPORARILY_UNAVAILABLE_THROTTLE_REQUESTS_FOR_ARG_MS_STATUS_ARG_F37192D9 = "Receiver {} 暂时不可用，对请求限流 {} ms。状态：{}"; + public static final String EXCEPTION_RECEIVER_ARG_REMAINED_TEMPORARILY_UNAVAILABLE_FOR_MORE_THAN_ARG_MS_PAUSE_REGULAR_RETRIES_AND_PROBE_EVERY_ARG_MS_C515DD97 = + "Receiver %s 持续暂时不可用超过 %d ms，暂停常规重试，改为每 %d ms 探测一次。"; public static final String MESSAGE_SUCCESSFULLY_TRANSFERRED_BATCHED_SCHEMA_EVENTS_BATCH_SIZE_ARG_CF2E881C = "成功传输批量的 schema 事件，batch 大小 {}。"; public static final String EXCEPTION_AUTO_CREATE_TREE_DATABASE_FAILED_ARG_STATUS_CODE_ARG_C6175C27 = 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 ce0b2c2233d4b..dc1db3386ddd4 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 @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException; 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.agent.task.subtask.PipeAbstractSinkSubtask; @@ -239,6 +240,8 @@ private void transferHeartbeatEvent(final PipeHeartbeatEvent event) { })) { return; } + } catch (final PipeRuntimeSinkNonReportTimeConfigurableException e) { + throw e; } catch (final Exception e) { throw new PipeConnectionException( String.format( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java index 9466543d1648e..6153d19e4441c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException; import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; @@ -643,6 +644,8 @@ public long consumeSchedulingDelayMs() { * @see PipeConnector#transfer(TsFileInsertionEvent) for more details. */ private void transferQueuedEventsIfNecessary(final boolean forced) { + throwIfReceiverProbeIsDelayed(); + if ((retryEventQueue.isEmpty() && retryTsFileQueue.isEmpty()) || (!forced && retryEventQueueEventCounter.getTabletInsertionEventCount() @@ -702,6 +705,8 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { } } + throwIfReceiverProbeIsDelayed(); + // Stop retrying if the execution time exceeds the threshold for better realtime performance if (System.currentTimeMillis() - retryStartTime > PipeConfig.getInstance().getPipeAsyncSinkMaxRetryExecutionTimeMsPerCall()) { @@ -849,21 +854,63 @@ public void waitIfReceiverTemporarilyUnavailable(final TEndPoint endPoint) { return; } - while (!isClosed.get()) { - final long waitTimeInMs = backoff.getRemainingWaitTimeInMs(); - if (waitTimeInMs <= 0) { - return; + while (!isClosed.get() && backoff.isActive()) { + if (backoff.isRetryMaxDurationExceeded()) { + final long probeDelayInMs = backoff.tryAcquireProbeAndGetDelayInMs(); + if (probeDelayInMs <= 0) { + return; + } + schedulingDelayMs.accumulateAndGet(probeDelayInMs, Math::max); + throw createReceiverProbeDelayException(endPointKey, backoff); } - try { - Thread.sleep(waitTimeInMs); - } catch (final InterruptedException e) { - Thread.currentThread().interrupt(); - return; + final long retryTimeInMs = backoff.reserveNextRetryTimeInMs(); + while (!isClosed.get() && backoff.isActive()) { + if (backoff.isRetryMaxDurationExceeded()) { + break; + } + + final long waitTimeInMs = retryTimeInMs - System.currentTimeMillis(); + if (waitTimeInMs <= 0) { + return; + } + + try { + Thread.sleep(Math.min(waitTimeInMs, 1000L)); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } } } } + private void throwIfReceiverProbeIsDelayed() { + for (final Map.Entry entry : + receiverBackoffMap.entrySet()) { + final long probeDelayInMs = entry.getValue().getRemainingProbeDelayInMs(); + if (probeDelayInMs <= 0) { + continue; + } + + schedulingDelayMs.accumulateAndGet(probeDelayInMs, Math::max); + throw createReceiverProbeDelayException(entry.getKey(), entry.getValue()); + } + } + + private static PipeRuntimeSinkNonReportTimeConfigurableException + createReceiverProbeDelayException( + final String endPointKey, final ReceiverTemporaryUnavailableBackoff backoff) { + return new PipeRuntimeSinkNonReportTimeConfigurableException( + String.format( + DataNodePipeMessages + .EXCEPTION_RECEIVER_ARG_REMAINED_TEMPORARILY_UNAVAILABLE_FOR_MORE_THAN_ARG_MS_PAUSE_REGULAR_RETRIES_AND_PROBE_EVERY_ARG_MS_C515DD97, + endPointKey, + backoff.getRetryMaxDurationInMs(), + backoff.getRetryProbeIntervalInMs()), + Long.MAX_VALUE); + } + public void recordReceiverStatus(final TEndPoint endPoint, final TSStatus status) { final String endPointKey = format(endPoint); if (Objects.isNull(endPointKey) || Objects.isNull(status)) { @@ -884,9 +931,17 @@ public void recordReceiverStatus(final TEndPoint endPoint, final TSStatus status status); } } else if (isSuccess(status)) { - final ReceiverTemporaryUnavailableBackoff backoff = receiverBackoffMap.get(endPointKey); - if (Objects.nonNull(backoff) && backoff.getRemainingWaitTimeInMs() <= 0) { - receiverBackoffMap.remove(endPointKey, backoff); + receiverBackoffMap.computeIfPresent( + endPointKey, + (key, backoff) -> { + if (!backoff.shouldResetOnSuccess()) { + return backoff; + } + backoff.markAvailable(); + return null; + }); + if (receiverBackoffMap.isEmpty()) { + schedulingDelayMs.set(0); } } } @@ -1097,23 +1152,90 @@ private static class ReceiverTemporaryUnavailableBackoff { private final long maxBackoffTimeInMs = Math.max(0, PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalMaxMs()); - private final AtomicLong currentBackoffTimeInMs = - new AtomicLong( - Math.min( - Math.max(1, PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalInitMs()), - maxBackoffTimeInMs)); - private final AtomicLong nextAvailableTimeInMs = new AtomicLong(0); - - private long markTemporarilyUnavailable() { - final long backoffTimeInMs = currentBackoffTimeInMs.get(); - nextAvailableTimeInMs.updateAndGet( - current -> Math.max(current, System.currentTimeMillis() + backoffTimeInMs)); - currentBackoffTimeInMs.updateAndGet(this::getNextBackoffTimeInMs); + private final long initialBackoffTimeInMs = + Math.min( + Math.max(1, PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalInitMs()), + maxBackoffTimeInMs); + private final long retryMaxDurationInMs = + PipeConfig.getInstance().getPipeAsyncSinkRetryMaxDurationMs(); + private final long retryProbeIntervalInMs = + Math.max(1, PipeConfig.getInstance().getPipeAsyncSinkRetryProbeIntervalMs()); + + private boolean active = false; + private long firstUnavailableTimeInMs = 0; + private long currentBackoffTimeInMs = initialBackoffTimeInMs; + private long failureBackoffUntilInMs = 0; + private long nextReservedRetryTimeInMs = 0; + private long nextProbeTimeInMs = 0; + + private synchronized long markTemporarilyUnavailable() { + final long currentTimeInMs = System.currentTimeMillis(); + if (!active) { + active = true; + firstUnavailableTimeInMs = currentTimeInMs; + currentBackoffTimeInMs = initialBackoffTimeInMs; + failureBackoffUntilInMs = 0; + nextReservedRetryTimeInMs = 0; + nextProbeTimeInMs = 0; + } + + final long backoffTimeInMs = currentBackoffTimeInMs; + failureBackoffUntilInMs = + Math.max(failureBackoffUntilInMs, safeAdd(currentTimeInMs, backoffTimeInMs)); + nextReservedRetryTimeInMs = Math.max(nextReservedRetryTimeInMs, failureBackoffUntilInMs); + currentBackoffTimeInMs = getNextBackoffTimeInMs(currentBackoffTimeInMs); return backoffTimeInMs; } - private long getRemainingWaitTimeInMs() { - return nextAvailableTimeInMs.get() - System.currentTimeMillis(); + private synchronized boolean isActive() { + return active; + } + + private synchronized boolean isRetryMaxDurationExceeded() { + return active + && retryMaxDurationInMs >= 0 + && System.currentTimeMillis() - firstUnavailableTimeInMs >= retryMaxDurationInMs; + } + + private synchronized long reserveNextRetryTimeInMs() { + final long currentTimeInMs = System.currentTimeMillis(); + final long retryTimeInMs = + Math.max(currentTimeInMs, Math.max(failureBackoffUntilInMs, nextReservedRetryTimeInMs)); + nextReservedRetryTimeInMs = safeAdd(retryTimeInMs, currentBackoffTimeInMs); + return retryTimeInMs; + } + + private synchronized long tryAcquireProbeAndGetDelayInMs() { + final long currentTimeInMs = System.currentTimeMillis(); + if (currentTimeInMs >= nextProbeTimeInMs) { + nextProbeTimeInMs = safeAdd(currentTimeInMs, retryProbeIntervalInMs); + return 0; + } + return nextProbeTimeInMs - currentTimeInMs; + } + + private synchronized long getRemainingProbeDelayInMs() { + return isRetryMaxDurationExceeded() + ? Math.max(0, nextProbeTimeInMs - System.currentTimeMillis()) + : 0; + } + + private synchronized boolean shouldResetOnSuccess() { + return active + && (isRetryMaxDurationExceeded() + || failureBackoffUntilInMs - System.currentTimeMillis() <= 0); + } + + private synchronized void markAvailable() { + active = false; + } + + private long getRetryMaxDurationInMs() { + return retryMaxDurationInMs; + } + + private long getRetryProbeIntervalInMs() { + return retryProbeIntervalInMs; } private long getNextBackoffTimeInMs(final long currentBackoffTimeInMs) { @@ -1124,5 +1246,9 @@ private long getNextBackoffTimeInMs(final long currentBackoffTimeInMs) { ? maxBackoffTimeInMs : currentBackoffTimeInMs << 1; } + + private static long safeAdd(final long left, final long right) { + return left >= Long.MAX_VALUE - right ? Long.MAX_VALUE : left + right; + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java index 53967d94a6b92..a3792bcb3daac 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferSliceReqBuilder; import org.apache.iotdb.db.i18n.DataNodePipeMessages; @@ -93,7 +94,7 @@ public void onError(final Exception exception) { * @param client the client used for data transfer * @param req the request containing transfer details * @return {@code true} if the transfer was initiated successfully, {@code false} if the connector - * is closed + * is closed or the receiver probe is delayed * @throws TException if an error occurs during the transfer */ protected boolean tryTransfer( @@ -107,7 +108,13 @@ protected boolean tryTransfer( if (returnFalseIfSinkIsClosed(client)) { return false; } - sink.waitIfReceiverTemporarilyUnavailable(client.getEndPoint()); + try { + sink.waitIfReceiverTemporarilyUnavailable(client.getEndPoint()); + } catch (final PipeRuntimeSinkNonReportTimeConfigurableException e) { + returnClientToPool(client); + onError(e); + return false; + } if (returnFalseIfSinkIsClosed(client)) { return false; } @@ -138,6 +145,22 @@ private boolean returnFalseIfSinkIsClosed(final AsyncPipeDataTransferServiceClie return true; } + private void returnClientToPool(final AsyncPipeDataTransferServiceClient client) { + client.setShouldReturnSelf(true); + client.returnSelf( + e -> { + if (e instanceof IllegalStateException) { + PipeLogger.log( + ignored -> + LOGGER.info(DataNodePipeMessages.ILLEGAL_STATE_WHEN_RETURN_THE_CLIENT_TO), + DataNodePipeMessages.ILLEGAL_STATE_WHEN_RETURN_THE_CLIENT_TO); + return true; + } + return false; + }); + this.client = null; + } + /** * @return {@code true} if all transmissions corresponding to the handler have been completed, * {@code false} otherwise @@ -281,6 +304,9 @@ private void fallbackToWholeRequest( return; } client.pipeTransfer(originalReq, this); + } catch (final PipeRuntimeSinkNonReportTimeConfigurableException e) { + returnClientToPool(client); + PipeTransferTrackableHandler.this.onError(e); } catch (final Exception e) { PipeTransferTrackableHandler.this.onError(e); } 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 c60b9c2c25185..8af86f1b128de 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 @@ -20,9 +20,11 @@ package org.apache.iotdb.db.pipe.agent.task.subtask.sink; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException; 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.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; 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; @@ -44,6 +46,7 @@ import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.mockito.Mockito.withSettings; @@ -266,6 +269,47 @@ public void testTransferExceptionUsesDisplayTaskID() throws Exception { } } + @Test + public void testHeartbeatPreservesReceiverProbeDelayException() throws Exception { + final long originalSleepIntervalInitMs = + CommonDescriptor.getInstance().getConfig().getPipeSinkSubtaskSleepIntervalInitMs(); + final long originalSleepIntervalMaxMs = + CommonDescriptor.getInstance().getConfig().getPipeSinkSubtaskSleepIntervalMaxMs(); + CommonDescriptor.getInstance().getConfig().setPipeSinkSubtaskSleepIntervalInitMs(1); + CommonDescriptor.getInstance().getConfig().setPipeSinkSubtaskSleepIntervalMaxMs(2); + + final PipeConnector connector = mock(PipeConnector.class); + final UnboundedBlockingPendingQueue pendingQueue = + mock(UnboundedBlockingPendingQueue.class); + when(pendingQueue.waitedPoll()).thenReturn(new PipeHeartbeatEvent(1, false)); + doThrow(new PipeRuntimeSinkNonReportTimeConfigurableException("probe delayed", Long.MAX_VALUE)) + .when(connector) + .transfer(any(Event.class)); + + final PipeSinkSubtask subtask = + new PipeSinkSubtask( + "PipeSinkSubtaskTest", + System.currentTimeMillis(), + "data_test", + "data_test", + 0, + pendingQueue, + connector); + + try { + Assert.assertTrue(subtask.executeOnce()); + verify(connector, never()).handshake(); + } finally { + subtask.close(); + CommonDescriptor.getInstance() + .getConfig() + .setPipeSinkSubtaskSleepIntervalInitMs(originalSleepIntervalInitMs); + CommonDescriptor.getInstance() + .getConfig() + .setPipeSinkSubtaskSleepIntervalMaxMs(originalSleepIntervalMaxMs); + } + } + private static class BlockingHandshakeConnector implements PipeConnector, PipeConnectorWithEventDiscard { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java index 8e0f780299881..49d1f617afa4d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandlerTest.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.IoTDBSinkRequestVersion; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType; import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferSliceReq; @@ -52,16 +53,28 @@ public class PipeTransferTrackableHandlerTest { private final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); private int originalRequestSliceThresholdBytes; + private long originalRetryMaxDurationMs; + private long originalRetryProbeIntervalMs; + private long originalSinkSubtaskSleepIntervalInitMs; + private long originalSinkSubtaskSleepIntervalMaxMs; @Before public void setUp() { originalRequestSliceThresholdBytes = commonConfig.getPipeSinkRequestSliceThresholdBytes(); + originalRetryMaxDurationMs = commonConfig.getPipeAsyncSinkRetryMaxDurationMs(); + originalRetryProbeIntervalMs = commonConfig.getPipeAsyncSinkRetryProbeIntervalMs(); + originalSinkSubtaskSleepIntervalInitMs = commonConfig.getPipeSinkSubtaskSleepIntervalInitMs(); + originalSinkSubtaskSleepIntervalMaxMs = commonConfig.getPipeSinkSubtaskSleepIntervalMaxMs(); commonConfig.setPipeSinkRequestSliceThresholdBytes(4); } @After public void tearDown() { commonConfig.setPipeSinkRequestSliceThresholdBytes(originalRequestSliceThresholdBytes); + commonConfig.setPipeAsyncSinkRetryMaxDurationMs(originalRetryMaxDurationMs); + commonConfig.setPipeAsyncSinkRetryProbeIntervalMs(originalRetryProbeIntervalMs); + commonConfig.setPipeSinkSubtaskSleepIntervalInitMs(originalSinkSubtaskSleepIntervalInitMs); + commonConfig.setPipeSinkSubtaskSleepIntervalMaxMs(originalSinkSubtaskSleepIntervalMaxMs); } @Test @@ -186,6 +199,69 @@ public void testTransferWaitsForReceiverBackoffAndRecordsStatus() throws Excepti Mockito.verify(sink).recordReceiverStatus(endPoint, status); } + @Test + public void testClientIsReturnedWhenReceiverProbeIsDelayed() throws Exception { + final IoTDBDataRegionAsyncSink sink = Mockito.mock(IoTDBDataRegionAsyncSink.class); + final AsyncPipeDataTransferServiceClient client = + Mockito.mock(AsyncPipeDataTransferServiceClient.class); + final TEndPoint endPoint = new TEndPoint("127.0.0.1", 6667); + final PipeRuntimeSinkNonReportTimeConfigurableException exception = + new PipeRuntimeSinkNonReportTimeConfigurableException("probe delayed", Long.MAX_VALUE); + Mockito.when(client.getEndPoint()).thenReturn(endPoint); + Mockito.doThrow(exception).when(sink).waitIfReceiverTemporarilyUnavailable(endPoint); + + final TestPipeTransferTrackableHandler handler = new TestPipeTransferTrackableHandler(sink); + + handler.transfer(client, createReq(1)); + Assert.assertEquals(1, handler.errorCount); + Mockito.verify(client).setShouldReturnSelf(true); + Mockito.verify(client).returnSelf(Mockito.any()); + Mockito.verify(client, Mockito.never()) + .pipeTransfer(Mockito.any(TPipeTransferReq.class), Mockito.any()); + } + + @Test + public void testReceiverRetriesAreSerialized() { + commonConfig.setPipeSinkSubtaskSleepIntervalInitMs(40); + commonConfig.setPipeSinkSubtaskSleepIntervalMaxMs(40); + commonConfig.setPipeAsyncSinkRetryMaxDurationMs(5000); + + final IoTDBDataRegionAsyncSink sink = new IoTDBDataRegionAsyncSink(); + final TEndPoint endPoint = new TEndPoint("127.0.0.1", 6667); + sink.recordReceiverStatus(endPoint, temporarilyUnavailableStatus()); + + final long startTimeInMs = System.currentTimeMillis(); + sink.waitIfReceiverTemporarilyUnavailable(endPoint); + sink.waitIfReceiverTemporarilyUnavailable(endPoint); + + Assert.assertTrue(System.currentTimeMillis() - startTimeInMs >= 60); + } + + @Test + public void testReceiverRetryFallsBackToSingleProbeAfterMaxDuration() { + commonConfig.setPipeAsyncSinkRetryMaxDurationMs(0); + commonConfig.setPipeAsyncSinkRetryProbeIntervalMs(1000); + + final IoTDBDataRegionAsyncSink sink = new IoTDBDataRegionAsyncSink(); + final TEndPoint endPoint = new TEndPoint("127.0.0.1", 6667); + sink.recordReceiverStatus(endPoint, temporarilyUnavailableStatus()); + + sink.waitIfReceiverTemporarilyUnavailable(endPoint); + Assert.assertThrows( + PipeRuntimeSinkNonReportTimeConfigurableException.class, + () -> sink.waitIfReceiverTemporarilyUnavailable(endPoint)); + Assert.assertTrue(sink.peekSchedulingDelayMs() > 0); + + sink.recordReceiverStatus( + endPoint, new TSStatus().setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + sink.waitIfReceiverTemporarilyUnavailable(endPoint); + } + + private static TSStatus temporarilyUnavailableStatus() { + return new TSStatus() + .setCode(TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode()); + } + private static TPipeTransferReq createReq(final int bodySize) { final byte[] body = new byte[bodySize]; for (int i = 0; i < body.length; ++i) { diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ConfigMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ConfigMessages.java index ff624744e232e..8ce3e9047886f 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ConfigMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ConfigMessages.java @@ -91,6 +91,10 @@ private ConfigMessages() {} public static final String LOG_PIPEASYNCSINKFORCEDRETRYTABLETEVENTQUEUESIZE_SET_ARG_8FDA7023 = "pipeAsyncSinkForcedRetryTabletEventQueueSize is set to {}."; public static final String LOG_PIPEASYNCSINKFORCEDRETRYTOTALEVENTQUEUESIZE_SET_ARG_92D6EACB = "pipeAsyncSinkForcedRetryTotalEventQueueSize is set to {}."; public static final String LOG_PIPEASYNCSINKMAXRETRYEXECUTIONTIMEMSPERCALL_SET_ARG_77E7B216 = "pipeAsyncSinkMaxRetryExecutionTimeMsPerCall is set to {}."; + public static final String LOG_PIPEASYNCSINKRETRYMAXDURATIONMS_IS_SET_TO_ARG_5058C99F = + "pipeAsyncSinkRetryMaxDurationMs is set to {}."; + public static final String LOG_PIPEASYNCSINKRETRYPROBEINTERVALMS_IS_SET_TO_ARG_A1E9AF45 = + "pipeAsyncSinkRetryProbeIntervalMs is set to {}."; public static final String LOG_PIPEASYNCSINKSELECTORNUMBER_SHOULD_GREATER_THAN_0_CONFIGURING_IT_NOT_CHANGE_EEB9793C = "pipeAsyncSinkSelectorNumber should be greater than 0, configuring it not to change."; public static final String LOG_PIPEASYNCSINKMAXCLIENTNUMBER_SHOULD_GREATER_THAN_0_CONFIGURING_IT_NOT_CHANGE_11EF47BF = " pipeAsyncSinkMaxClientNumber should be greater than 0, configuring it not to change."; public static final String LOG_PIPEASYNCSINKMAXTSFILECLIENTNUMBER_SHOULD_GREATER_THAN_0_CONFIGURING_IT_NOT_CHANGE_AC812FE2 = "pipeAsyncSinkMaxTsFileClientNumber should be greater than 0, configuring it not to change."; diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index 3ac85af31213b..aa04fcafbd9df 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -145,6 +145,10 @@ public final class PipeMessages { "PipeAsyncSinkForcedRetryTotalEventQueueSizeThreshold: {}"; public static final String CONFIG_PIPE_ASYNC_SINK_MAX_RETRY_EXECUTION_TIME_MS_PER_CALL = "PipeAsyncSinkMaxRetryExecutionTimeMsPerCall: {}"; + public static final String CONFIG_PIPE_ASYNC_SINK_RETRY_MAX_DURATION_MS = + "PipeAsyncSinkRetryMaxDurationMs: {}"; + public static final String CONFIG_PIPE_ASYNC_SINK_RETRY_PROBE_INTERVAL_MS = + "PipeAsyncSinkRetryProbeIntervalMs: {}"; public static final String CONFIG_PIPE_ASYNC_SINK_SELECTOR_NUMBER = "PipeAsyncSinkSelectorNumber: {}"; public static final String CONFIG_PIPE_ASYNC_SINK_MAX_CLIENT_NUMBER = diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ConfigMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ConfigMessages.java index 5217d77bf1b26..f8b810fde13df 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ConfigMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ConfigMessages.java @@ -91,6 +91,10 @@ private ConfigMessages() {} public static final String LOG_PIPEASYNCSINKFORCEDRETRYTABLETEVENTQUEUESIZE_SET_ARG_8FDA7023 = "pipeAsyncSinkForcedRetryTabletEventQueueSize 已设置为 {}。"; public static final String LOG_PIPEASYNCSINKFORCEDRETRYTOTALEVENTQUEUESIZE_SET_ARG_92D6EACB = "pipeAsyncSinkForcedRetryTotalEventQueueSize 已设置为 {}。"; public static final String LOG_PIPEASYNCSINKMAXRETRYEXECUTIONTIMEMSPERCALL_SET_ARG_77E7B216 = "pipeAsyncSinkMaxRetryExecutionTimeMsPerCall 已设置为 {}。"; + public static final String LOG_PIPEASYNCSINKRETRYMAXDURATIONMS_IS_SET_TO_ARG_5058C99F = + "pipeAsyncSinkRetryMaxDurationMs 已设置为 {}。"; + public static final String LOG_PIPEASYNCSINKRETRYPROBEINTERVALMS_IS_SET_TO_ARG_A1E9AF45 = + "pipeAsyncSinkRetryProbeIntervalMs 已设置为 {}。"; public static final String LOG_PIPEASYNCSINKSELECTORNUMBER_SHOULD_GREATER_THAN_0_CONFIGURING_IT_NOT_CHANGE_EEB9793C = "pipeAsyncSinkSelectorNumber 应大于 0，配置保持不变。"; public static final String LOG_PIPEASYNCSINKMAXCLIENTNUMBER_SHOULD_GREATER_THAN_0_CONFIGURING_IT_NOT_CHANGE_11EF47BF = "pipeAsyncSinkMaxClientNumber 应大于 0，配置保持不变。"; public static final String LOG_PIPEASYNCSINKMAXTSFILECLIENTNUMBER_SHOULD_GREATER_THAN_0_CONFIGURING_IT_NOT_CHANGE_AC812FE2 = "pipeAsyncSinkMaxTsFileClientNumber 应大于 0，配置保持不变。"; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index 516a890852dd2..9c808eecbf830 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -144,6 +144,10 @@ public final class PipeMessages { "PipeAsyncSinkForcedRetryTotalEventQueueSizeThreshold: {}"; public static final String CONFIG_PIPE_ASYNC_SINK_MAX_RETRY_EXECUTION_TIME_MS_PER_CALL = "PipeAsyncSinkMaxRetryExecutionTimeMsPerCall: {}"; + public static final String CONFIG_PIPE_ASYNC_SINK_RETRY_MAX_DURATION_MS = + "PipeAsyncSinkRetryMaxDurationMs: {}"; + public static final String CONFIG_PIPE_ASYNC_SINK_RETRY_PROBE_INTERVAL_MS = + "PipeAsyncSinkRetryProbeIntervalMs: {}"; public static final String CONFIG_PIPE_ASYNC_SINK_SELECTOR_NUMBER = "PipeAsyncSinkSelectorNumber: {}"; public static final String CONFIG_PIPE_ASYNC_SINK_MAX_CLIENT_NUMBER = diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index c811ae9f5f904..a52112283009c 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -292,6 +292,8 @@ public class CommonConfig { private int pipeAsyncSinkForcedRetryTabletEventQueueSize = 20; private int pipeAsyncSinkForcedRetryTotalEventQueueSize = 30; private long pipeAsyncSinkMaxRetryExecutionTimeMsPerCall = 500; + private long pipeAsyncSinkRetryMaxDurationMs = 60 * 1000L; + private long pipeAsyncSinkRetryProbeIntervalMs = 30 * 1000L; private int pipeAsyncSinkSelectorNumber = Math.max(4, Runtime.getRuntime().availableProcessors() / 2); private int pipeAsyncSinkMaxClientNumber = @@ -1326,6 +1328,34 @@ public long getPipeAsyncSinkMaxRetryExecutionTimeMsPerCall() { return pipeAsyncSinkMaxRetryExecutionTimeMsPerCall; } + public void setPipeAsyncSinkRetryMaxDurationMs(long pipeAsyncSinkRetryMaxDurationMs) { + if (this.pipeAsyncSinkRetryMaxDurationMs == pipeAsyncSinkRetryMaxDurationMs) { + return; + } + this.pipeAsyncSinkRetryMaxDurationMs = pipeAsyncSinkRetryMaxDurationMs; + logger.info( + ConfigMessages.LOG_PIPEASYNCSINKRETRYMAXDURATIONMS_IS_SET_TO_ARG_5058C99F, + pipeAsyncSinkRetryMaxDurationMs); + } + + public long getPipeAsyncSinkRetryMaxDurationMs() { + return pipeAsyncSinkRetryMaxDurationMs; + } + + public void setPipeAsyncSinkRetryProbeIntervalMs(long pipeAsyncSinkRetryProbeIntervalMs) { + if (this.pipeAsyncSinkRetryProbeIntervalMs == pipeAsyncSinkRetryProbeIntervalMs) { + return; + } + this.pipeAsyncSinkRetryProbeIntervalMs = Math.max(1, pipeAsyncSinkRetryProbeIntervalMs); + logger.info( + ConfigMessages.LOG_PIPEASYNCSINKRETRYPROBEINTERVALMS_IS_SET_TO_ARG_A1E9AF45, + this.pipeAsyncSinkRetryProbeIntervalMs); + } + + public long getPipeAsyncSinkRetryProbeIntervalMs() { + return pipeAsyncSinkRetryProbeIntervalMs; + } + public int getPipeAsyncSinkSelectorNumber() { return pipeAsyncSinkSelectorNumber; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java index dd2edcb57ee13..6e85519a85f68 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java @@ -220,6 +220,14 @@ public long getPipeAsyncSinkMaxRetryExecutionTimeMsPerCall() { return COMMON_CONFIG.getPipeAsyncSinkMaxRetryExecutionTimeMsPerCall(); } + public long getPipeAsyncSinkRetryMaxDurationMs() { + return COMMON_CONFIG.getPipeAsyncSinkRetryMaxDurationMs(); + } + + public long getPipeAsyncSinkRetryProbeIntervalMs() { + return COMMON_CONFIG.getPipeAsyncSinkRetryProbeIntervalMs(); + } + public int getPipeAsyncSinkSelectorNumber() { return COMMON_CONFIG.getPipeAsyncSinkSelectorNumber(); } @@ -622,6 +630,12 @@ public void printAllConfigs() { LOGGER.info( PipeMessages.CONFIG_PIPE_ASYNC_SINK_MAX_RETRY_EXECUTION_TIME_MS_PER_CALL, getPipeAsyncSinkMaxRetryExecutionTimeMsPerCall()); + LOGGER.info( + PipeMessages.CONFIG_PIPE_ASYNC_SINK_RETRY_MAX_DURATION_MS, + getPipeAsyncSinkRetryMaxDurationMs()); + LOGGER.info( + PipeMessages.CONFIG_PIPE_ASYNC_SINK_RETRY_PROBE_INTERVAL_MS, + getPipeAsyncSinkRetryProbeIntervalMs()); LOGGER.info( PipeMessages.CONFIG_PIPE_ASYNC_SINK_SELECTOR_NUMBER, getPipeAsyncSinkSelectorNumber()); LOGGER.info( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java index 780d6b3252a07..7909ef89e80d9 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java @@ -403,6 +403,20 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr properties.getProperty( "pipe_async_connector_max_retry_execution_time_ms_per_call", String.valueOf(config.getPipeAsyncSinkMaxRetryExecutionTimeMsPerCall()))))); + config.setPipeAsyncSinkRetryMaxDurationMs( + Long.parseLong( + Optional.ofNullable(properties.getProperty("pipe_async_sink_retry_max_duration_ms")) + .orElse( + properties.getProperty( + "pipe_async_connector_retry_max_duration_ms", + String.valueOf(config.getPipeAsyncSinkRetryMaxDurationMs()))))); + config.setPipeAsyncSinkRetryProbeIntervalMs( + Long.parseLong( + Optional.ofNullable(properties.getProperty("pipe_async_sink_retry_probe_interval_ms")) + .orElse( + properties.getProperty( + "pipe_async_connector_retry_probe_interval_ms", + String.valueOf(config.getPipeAsyncSinkRetryProbeIntervalMs()))))); config.setPipeAsyncSinkForcedRetryTsFileEventQueueSize( Integer.parseInt( Optional.ofNullable(