jt2594838 commented on code in PR #17928:
URL: https://github.com/apache/iotdb/pull/17928#discussion_r3411644125
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java:
##########
@@ -100,28 +104,40 @@ protected boolean tryTransfer(
}
// track handler before checking if connector is closed
sink.trackHandler(this);
- if (sink.isClosed()) {
- clearEventsReferenceCount();
- sink.eliminateHandler(this, true);
- client.setShouldReturnSelf(true);
- client.returnSelf(
- (e) -> {
- if (e instanceof IllegalStateException) {
- PipeLogger.log(
- ignored ->
-
LOGGER.info(DataNodePipeMessages.ILLEGAL_STATE_WHEN_RETURN_THE_CLIENT_TO),
- "Illegal state when return the client to object pool, maybe
the pool is already cleared. Will ignore.");
- return true;
- }
- return false;
- });
- this.client = null;
+ if (returnFalseIfSinkIsClosed(client)) {
+ return false;
+ }
+ sink.waitIfReceiverTemporarilyUnavailable(client.getEndPoint());
+ if (returnFalseIfSinkIsClosed(client)) {
return false;
}
doTransfer(client, req);
return true;
}
+ private boolean returnFalseIfSinkIsClosed(final
AsyncPipeDataTransferServiceClient client) {
+ if (!sink.isClosed()) {
+ return false;
+ }
+
+ clearEventsReferenceCount();
+ sink.eliminateHandler(this, true);
+ client.setShouldReturnSelf(true);
+ client.returnSelf(
+ (e) -> {
+ if (e instanceof IllegalStateException) {
+ PipeLogger.log(
+ ignored ->
+
LOGGER.info(DataNodePipeMessages.ILLEGAL_STATE_WHEN_RETURN_THE_CLIENT_TO),
+ "Illegal state when return the client to object pool, maybe
the pool is already cleared. Will ignore.");
Review Comment:
i18n
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java:
##########
@@ -931,4 +1014,37 @@ public void setBatchEventSizeHistogram(Histogram
eventSizeHistogram) {
tabletBatchBuilder.setEventSizeHistogram(eventSizeHistogram);
}
}
+
+ private static class ReceiverTemporaryUnavailableBackoff {
+
+ private final long maxBackoffTimeInMs =
+ Math.max(0,
PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalMaxMs());
+ private final AtomicLong currentBackoffTimeInMs =
+ new AtomicLong(
+ Math.min(
+ Math.max(0,
PipeConfig.getInstance().getPipeSinkSubtaskSleepIntervalInitMs()),
Review Comment:
0 -> 1?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
