This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 8b107926d12 Fix unbounded async pipe sink retries (#18222) (#18225)
8b107926d12 is described below
commit 8b107926d12505a71f1d4d002a3bcdf6b1dd617a
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 16 19:06:49 2026 +0800
Fix unbounded async pipe sink retries (#18222) (#18225)
---
.../agent/task/subtask/sink/PipeSinkSubtask.java | 3 +
.../thrift/async/IoTDBDataRegionAsyncSink.java | 173 +++++++++++++++++----
.../handler/PipeTransferTrackableHandler.java | 29 +++-
.../task/subtask/sink/PipeSinkSubtaskTest.java | 46 ++++++
.../handler/PipeTransferTrackableHandlerTest.java | 74 +++++++++
.../apache/iotdb/commons/conf/CommonConfig.java | 27 ++++
.../iotdb/commons/pipe/config/PipeConfig.java | 10 ++
.../iotdb/commons/pipe/config/PipeDescriptor.java | 14 ++
8 files changed, 346 insertions(+), 30 deletions(-)
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 b8c7423c232..eed4b729ed2 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;
@@ -170,6 +171,8 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
})) {
return;
}
+ } catch (final PipeRuntimeSinkNonReportTimeConfigurableException e) {
+ throw e;
} catch (final Exception e) {
throw new PipeConnectionException(
"PipeConnector: "
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 abc10f4fcc2..13589527b37 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
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
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;
@@ -85,7 +86,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_ENABLE_SEND_TSFILE_LIMIT;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_ENABLE_SEND_TSFILE_LIMIT_DEFAULT_VALUE;
@@ -539,6 +539,8 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
* @see PipeConnector#transfer(TsFileInsertionEvent) for more details.
*/
private void transferQueuedEventsIfNecessary(final boolean forced) {
+ throwIfReceiverProbeIsDelayed();
+
if ((retryEventQueue.isEmpty() && retryTsFileQueue.isEmpty())
|| (!forced
&& retryEventQueueEventCounter.getTabletInsertionEventCount()
@@ -600,6 +602,8 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
}
}
+ throwIfReceiverProbeIsDelayed();
+
// Stop retrying if the execution time exceeds the threshold for better
realtime performance
if (System.currentTimeMillis() - retryStartTime
>
PipeConfig.getInstance().getPipeAsyncSinkMaxRetryExecutionTimeMsPerCall()) {
@@ -747,21 +751,58 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
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;
+ }
+ 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;
+ }
+
+ throw createReceiverProbeDelayException(entry.getKey(),
entry.getValue());
+ }
+ }
+
+ private static PipeRuntimeSinkNonReportTimeConfigurableException
+ createReceiverProbeDelayException(
+ final String endPointKey, final ReceiverTemporaryUnavailableBackoff
backoff) {
+ return new PipeRuntimeSinkNonReportTimeConfigurableException(
+ String.format(
+ "Receiver %s remained temporarily unavailable for more than %d ms,
pause regular retries and probe every %d ms.",
+ 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)) {
@@ -781,10 +822,15 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
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;
+ });
}
}
@@ -994,23 +1040,90 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
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) {
@@ -1021,5 +1134,9 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
? 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 6c72335b1d2..edae06ac65f 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.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
@@ -92,7 +93,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(
@@ -106,7 +107,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;
}
@@ -136,6 +143,21 @@ public abstract class PipeTransferTrackableHandler
return true;
}
+ private void returnClientToPool(final AsyncPipeDataTransferServiceClient
client) {
+ client.setShouldReturnSelf(true);
+ client.returnSelf(
+ e -> {
+ if (e instanceof IllegalStateException) {
+ PipeLogger.log(
+ LOGGER::info,
+ "Illegal state when return the client to object pool, maybe
the pool is already cleared. Will ignore.");
+ return true;
+ }
+ return false;
+ });
+ this.client = null;
+ }
+
/**
* @return {@code true} if all transmissions corresponding to the handler
have been completed,
* {@code false} otherwise
@@ -276,6 +298,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 7d30078eaa8..6f14de41471 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;
@@ -39,9 +41,13 @@ 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.never;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import static org.mockito.Mockito.withSettings;
public class PipeSinkSubtaskTest {
@@ -221,6 +227,46 @@ 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",
+ 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..61218d112c8 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,67 @@ 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));
+ 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/java/org/apache/iotdb/commons/conf/CommonConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 91fa09daac1..189c9d0e9a8 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
@@ -282,6 +282,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 =
@@ -1175,6 +1177,31 @@ public class CommonConfig {
return pipeAsyncSinkMaxRetryExecutionTimeMsPerCall;
}
+ public void setPipeAsyncSinkRetryMaxDurationMs(long
pipeAsyncSinkRetryMaxDurationMs) {
+ if (this.pipeAsyncSinkRetryMaxDurationMs ==
pipeAsyncSinkRetryMaxDurationMs) {
+ return;
+ }
+ this.pipeAsyncSinkRetryMaxDurationMs = pipeAsyncSinkRetryMaxDurationMs;
+ logger.info("pipeAsyncSinkRetryMaxDurationMs is set to {}.",
pipeAsyncSinkRetryMaxDurationMs);
+ }
+
+ public long getPipeAsyncSinkRetryMaxDurationMs() {
+ return pipeAsyncSinkRetryMaxDurationMs;
+ }
+
+ public void setPipeAsyncSinkRetryProbeIntervalMs(long
pipeAsyncSinkRetryProbeIntervalMs) {
+ if (this.pipeAsyncSinkRetryProbeIntervalMs ==
pipeAsyncSinkRetryProbeIntervalMs) {
+ return;
+ }
+ this.pipeAsyncSinkRetryProbeIntervalMs = Math.max(1,
pipeAsyncSinkRetryProbeIntervalMs);
+ logger.info(
+ "pipeAsyncSinkRetryProbeIntervalMs is set to {}.",
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 16250dc1e8a..a0456f7a5c4 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
@@ -223,6 +223,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();
}
@@ -593,6 +601,8 @@ public class PipeConfig {
LOGGER.info(
"PipeAsyncSinkMaxRetryExecutionTimeMsPerCall: {}",
getPipeAsyncSinkMaxRetryExecutionTimeMsPerCall());
+ LOGGER.info("PipeAsyncSinkRetryMaxDurationMs: {}",
getPipeAsyncSinkRetryMaxDurationMs());
+ LOGGER.info("PipeAsyncSinkRetryProbeIntervalMs: {}",
getPipeAsyncSinkRetryProbeIntervalMs());
LOGGER.info("PipeAsyncSinkSelectorNumber: {}",
getPipeAsyncSinkSelectorNumber());
LOGGER.info("PipeAsyncSinkMaxClientNumber: {}",
getPipeAsyncSinkMaxClientNumber());
LOGGER.info("PipeAsyncSinkMaxTsFileClientNumber: {}",
getPipeAsyncSinkMaxTsFileClientNumber());
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 2cf877db505..b9115769e11 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
@@ -411,6 +411,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(
