gosonzhang commented on code in PR #11144:
URL: https://github.com/apache/inlong/pull/11144#discussion_r1768260102
##########
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java:
##########
@@ -191,11 +195,93 @@ 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 < ConfigConstants.SENDER_MAX_RETRY) {
+ sendResult = sendOperation.apply(currentSender);
+ if (sendResult != null && sendResult.equals(SendResult.OK)) {
+ currentSender.setSendSuccess(true);
+ return sendResult;
+ }
+ currentSender.setSendSuccess(false);
+ // try to get success sender
+ DefaultMessageSender randomSuccessSender =
getRandomSuccessSender();
+ if (randomSuccessSender != null) {
+ currentSender = randomSuccessSender;
+ } 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 < ConfigConstants.SENDER_MAX_RETRY) {
+ 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 =
getRandomSuccessSender();
+ if (randomSuccessSender != null) {
+ currentSender = randomSuccessSender;
+ } else {
+ break;
+ }
+ attempts++;
+ }
+
+ return sendIndexResult;
+ }
+
+ private DefaultMessageSender getRandomSuccessSender() {
+ List<Integer> keys = new ArrayList<>(CACHE_SENDER.keySet());
+ if (keys.isEmpty()) {
+ return null;
+ }
+ int attempts = 0;
+ int maxAttempts = keys.size();
+ // choose sending success MessageSender randomly
+ while (attempts < maxAttempts) {
+ int randomIndex = ThreadLocalRandom.current().nextInt(keys.size());
Review Comment:
The senders are saved according to different clusterIds and cannot be
obtained randomly.
--
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
