Skip to content

Pipe: throttle prolonged async sink retries #18222

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 1 commit into from
Jul 16, 2026
Merged

Pipe: throttle prolonged async sink retries #18222

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Expand Down
2 changes: 2 additions & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Expand Down
3 changes: 3 additions & 0 deletions ...anode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand Down
178 changes: 152 additions & 26 deletions ...in/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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()) {
Expand Down Expand Up @@ -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<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)) {
Expand All @@ -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);
}
}
}
Expand Down Expand Up @@ -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) {
Expand All @@ -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;
}
}
}
30 changes: 28 additions & 2 deletions ...apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand All @@ -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;
}
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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);
}
Expand Down
Loading
Loading