emptyOVO commented on code in PR #11144:
URL: https://github.com/apache/inlong/pull/11144#discussion_r1775305053
##########
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java:
##########
@@ -191,11 +195,80 @@ public String getSDKVersion() {
return ConfigConstants.PROXY_SDK_VERSION;
}
+ public void setSendSuccess(boolean sendSuccess) {
+ this.sendSuccess = sendSuccess;
+ }
+
+ public boolean isSendSuccess() {
+ return sendSuccess;
+ }
+
+ private SendResult retryWhenSendMessageFail(Function<DefaultMessageSender,
SendResult> sendOperation,
+ DefaultMessageSender initialSender) {
+ int attempts = 0;
+ SendResult sendResult = null;
+ DefaultMessageSender currentSender = initialSender;
+ while (attempts < this.senderMaxRetry) {
+ sendResult = sendOperation.apply(currentSender);
+ if (sendResult != null && sendResult.equals(SendResult.OK)) {
+ currentSender.setSendSuccess(true);
+ return sendResult;
+ }
+ currentSender.setSendSuccess(false);
+ // try to get another success sender
+ DefaultMessageSender anotherSender =
getAnotherSuccessSender(currentSender);
+ if (anotherSender != null) {
+ currentSender = anotherSender;
+ } else {
+ break;
+ }
+ attempts++;
+ }
+
+ return sendResult;
+ }
+
+ private String
retryWhenSendMessageIndexFail(Function<DefaultMessageSender, String>
sendOperation,
+ DefaultMessageSender initialSender) {
+ int attempts = 0;
+ String sendIndexResult = null;
+ DefaultMessageSender currentSender = initialSender;
+ while (attempts < this.senderMaxRetry) {
+ sendIndexResult = sendOperation.apply(currentSender);
+ if (sendIndexResult != null &&
sendIndexResult.startsWith(SendResult.OK.toString())) {
+ currentSender.setSendSuccess(true);
+ return sendIndexResult;
+ }
+ currentSender.setSendSuccess(false);
+ // try to get success sender
+ DefaultMessageSender randomSuccessSender =
getAnotherSuccessSender(currentSender);
+ if (randomSuccessSender != null) {
+ currentSender = randomSuccessSender;
+ } else {
+ break;
+ }
+ attempts++;
+ }
+
+ return sendIndexResult;
+ }
+
+ private DefaultMessageSender getAnotherSuccessSender(DefaultMessageSender
currentSender) {
+ for (DefaultMessageSender sender : CACHE_SENDER.values()) {
Review Comment:
done
--
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: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
