This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5b311f6ef9e Fix unbounded async pipe sink retries (#18222)
5b311f6ef9e is described below
commit 5b311f6ef9e352b9315fce01bf34400701432e50
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 16 15:08:38 2026 +0800
Fix unbounded async pipe sink retries (#18222)
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../agent/task/subtask/sink/PipeSinkSubtask.java | 3 +
.../thrift/async/IoTDBDataRegionAsyncSink.java | 178 ++++++++++++++++++---
.../handler/PipeTransferTrackableHandler.java | 30 +++-
.../task/subtask/sink/PipeSinkSubtaskTest.java | 44 +++++
.../handler/PipeTransferTrackableHandlerTest.java | 76 +++++++++
.../apache/iotdb/commons/i18n/ConfigMessages.java | 4 +
.../apache/iotdb/commons/i18n/PipeMessages.java | 4 +
.../apache/iotdb/commons/i18n/ConfigMessages.java | 4 +
.../apache/iotdb/commons/i18n/PipeMessages.java | 4 +
.../apache/iotdb/commons/conf/CommonConfig.java | 30 ++++
.../iotdb/commons/pipe/config/PipeConfig.java | 14 ++
.../iotdb/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 af39e7b4b42..bfd57672ee3 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 @@ public final class 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 138f0c752db..2b97336eaba 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 @@ public final class 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 ce0b2c2233d..dc1db3386dd 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 @@ 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.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 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
})) {
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 9466543d164..6153d19e444 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.common.rpc.thrift.TSStatus;
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 class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
* @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 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
}
}
+ 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 class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
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<String, ReceiverTemporaryUnavailableBackoff> 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 class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
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 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
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 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
? 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 53967d94a6b..a3792bcb3da 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 @@ package
org.apache.iotdb.db.pipe.sink.protocol.thrift.async.handler;
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 abstract class PipeTransferTrackableHandler
* @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 @@ public abstract class PipeTransferTrackableHandler
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 @@ public abstract class PipeTransferTrackableHandler
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 @@ public abstract class PipeTransferTrackableHandler
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 c60b9c2c251..8af86f1b128 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.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.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.withSettings;
@@ -266,6 +269,47 @@ public class PipeSinkSubtaskTest {
}
}
+ @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<Event> 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 8e0f7802998..49d1f617afa 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.common.rpc.thrift.TSStatus;
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 class PipeTransferTrackableHandlerTest {
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 ff624744e23..8ce3e904788 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 @@ public final class 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 3ac85af3121..aa04fcafbd9 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 5217d77bf1b..f8b810fde13 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 @@ public final class 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 516a890852d..9c808eecbf8 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 c811ae9f5f9..a5211228300 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 class CommonConfig {
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 dd2edcb57ee..6e85519a85f 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 class PipeConfig {
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 class PipeConfig {
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 780d6b3252a..7909ef89e80 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 class PipeDescriptor {
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(
