lincoln-lil commented on code in PR #19983:
URL: https://github.com/apache/flink/pull/19983#discussion_r908426718
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java:
##########
@@ -197,11 +248,12 @@ public void processElement(StreamRecord<IN> record)
throws Exception {
// add element first to the queue
final ResultFuture<OUT> entry = addToWorkQueue(element);
- final ResultHandler resultHandler = new ResultHandler(element, entry);
+ final RetryableResultHandlerDelegator resultHandler =
Review Comment:
I've tried this way, but the code seems a little redundant:
```
if (retryEnabled) {
final RetryableResultHandlerDelegator resultHandler =
new RetryableResultHandlerDelegator(element, entry,
getProcessingTimeService());
// register a timeout for the entry
assert timeout > 0L;
resultHandler.registerTimeout(timeout);
userFunction.asyncInvoke(element.getValue(), resultHandler);
} else {
final ResultHandler resultHandler = new ResultHandler(element,
entry);
// register a timeout for the entry if timeout is configured
if (timeout > 0L) {
resultHandler.registerTimeout(getProcessingTimeService(),
timeout);
}
userFunction.asyncInvoke(element.getValue(), resultHandler);
}
```
So use the unified RetryableResultHandlerDelegator to deal with two branches
according to retryEnabled, and this function call 'should' be inlined by jvm.
WDYT?
--
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]
