[Issue #528] enhance Retry support for HTTP and TCP processors. (#529)

* [Issue #337] Fix HttpSubscriber startup issue * [Issue #337] test commit * [Issue #337] revert test commit * [Issue #337] Enhance Http Demo Subscriber by using ExecutorService, CountDownLatch and PreDestroy hook * [Issue #337] Enhance Http Demo Subscriber by using ExecutorService, CountDownLatch and PreDestroy hook * [Issue #337] Address code review comment for Subscriber Demo App * [Issue #528] enhance Retry support for HTTP and TCP processors. * [Issue #528] fixing checkstyle issue * [Issue #528] Fix taskExecuteTime in TCP ClientGroupWrapper UpStreamMsgContext Co-authored-by: j00441484 <[email protected]>
apache · Sep 30, 2021 · e1e9eb0 · e1e9eb0
1 parent dcd409d
commit e1e9eb0
Show file tree

Hide file tree

Showing 11 changed files with 397 additions and 299 deletions.
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
@@ -44,7 +44,7 @@
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.group.ClientSessionGroupMapping;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.rebalance.EventMeshRebalanceService;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.rebalance.EventmeshRebalanceImpl;
-import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.push.retry.EventMeshTcpRetryer;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.retry.EventMeshTcpRetryer;
 import org.apache.eventmesh.runtime.metrics.tcp.EventMeshTcpMonitor;
 import org.apache.eventmesh.runtime.registry.Registry;
 import org.apache.eventmesh.runtime.util.EventMeshThreadFactoryImpl;

diff --git a/.../org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java b/.../org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
@@ -245,6 +245,8 @@ public void onException(OnExceptionContext context) {
                             SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getRetCode(),
                                     EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(context.getException(), 2)));
                     asyncContext.onComplete(err, handler);
+
+                    eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
                     long endTime = System.currentTimeMillis();
                     eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
                     eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgCost(endTime - startTime);
@@ -261,6 +263,8 @@ public void onException(OnExceptionContext context) {
                     SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getRetCode(),
                             EventMeshRetCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(ex, 2)));
             asyncContext.onComplete(err);
+
+            eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
             long endTime = System.currentTimeMillis();
             messageLogger.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
                     endTime - startTime,

diff --git a/...a/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java b/...a/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
@@ -254,6 +254,8 @@ public void onException(Throwable e) {
                             SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_WAITING_RR_MSG_ERR.getRetCode(),
                                     EventMeshRetCode.EVENTMESH_WAITING_RR_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2)));
                     asyncContext.onComplete(err, handler);
+
+                    eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
                     messageLogger.error("message|mq2eventMesh|RSP|SYNC|rrCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
                             System.currentTimeMillis() - startTime,
                             sendMessageRequestBody.getTopic(),
@@ -267,6 +269,8 @@ public void onException(Throwable e) {
                     SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getRetCode(),
                             EventMeshRetCode.EVENTMESH_SEND_SYNC_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(ex, 2)));
             asyncContext.onComplete(err);
+
+            eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
             long endTime = System.currentTimeMillis();
             eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
             eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgCost(endTime - startTime);

diff --git a/.../java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java b/.../java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
@@ -17,26 +17,8 @@
 
 package org.apache.eventmesh.runtime.core.protocol.tcp.client.group;
 
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
 import com.alibaba.fastjson.JSON;
-
-import io.openmessaging.api.AsyncConsumeContext;
-import io.openmessaging.api.AsyncMessageListener;
-import io.openmessaging.api.Message;
-import io.openmessaging.api.OnExceptionContext;
-import io.openmessaging.api.SendCallback;
-import io.openmessaging.api.SendResult;
-
+import io.openmessaging.api.*;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.eventmesh.api.EventMeshAction;
@@ -53,14 +35,20 @@
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.group.dispatch.DownstreamDispatchStrategy;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.push.DownStreamMsgContext;
-import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.push.retry.EventMeshTcpRetryer;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.retry.EventMeshTcpRetryer;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.send.UpStreamMsgContext;
 import org.apache.eventmesh.runtime.metrics.tcp.EventMeshTcpMonitor;
 import org.apache.eventmesh.runtime.util.EventMeshUtil;
 import org.apache.eventmesh.runtime.util.HttpTinyClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
 public class ClientGroupWrapper {
 
     public static Logger logger = LoggerFactory.getLogger(ClientGroupWrapper.class);
@@ -752,7 +740,9 @@ private void sendMsgBackToBroker(Message msg, String bizSeqNo) throws Exception
             String topic = msg.getSystemProperties(Constants.PROPERTY_MESSAGE_DESTINATION);
             logger.warn("send msg back to broker, bizSeqno:{}, topic:{}", bizSeqNo, topic);
 
-            send(new UpStreamMsgContext(null, null, msg), new SendCallback() {
+            long startTime = System.currentTimeMillis();
+            long taskExcuteTime = startTime;
+            send(new UpStreamMsgContext(null, msg, null, startTime, taskExcuteTime), new SendCallback() {
                 @Override
                 public void onSuccess(SendResult sendResult) {
                     logger.info("consumerGroup:{} consume fail, sendMessageBack success, bizSeqno:{}, topic:{}", consumerGroup, bizSeqNo, topic);

diff --git a/.../apache/eventmesh/runtime/core/protocol/tcp/client/session/push/DownStreamMsgContext.java b/.../apache/eventmesh/runtime/core/protocol/tcp/client/session/push/DownStreamMsgContext.java
@@ -17,45 +17,36 @@
 
 package org.apache.eventmesh.runtime.core.protocol.tcp.client.session.push;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Delayed;
-import java.util.concurrent.TimeUnit;
-
 import io.openmessaging.api.Message;
-
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.time.DateFormatUtils;
 import org.apache.eventmesh.api.AbstractContext;
 import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.common.protocol.SubscriptionItem;
+import org.apache.eventmesh.common.protocol.SubscriptionMode;
 import org.apache.eventmesh.runtime.constants.EventMeshConstants;
 import org.apache.eventmesh.runtime.core.plugin.MQConsumerWrapper;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.retry.RetryContext;
+import org.apache.eventmesh.runtime.util.EventMeshUtil;
 import org.apache.eventmesh.runtime.util.ServerGlobal;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
 
-public class DownStreamMsgContext implements Delayed {
+public class DownStreamMsgContext extends RetryContext {
 
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
-    public String seq;
-
-    public Message msgExt;
-
     public Session session;
 
     public AbstractContext consumeConcurrentlyContext;
 
     public MQConsumerWrapper consumer;
 
-    public int retryTimes;
-
     public SubscriptionItem subscriptionItem;
 
-    private long executeTime;
-
     public long lastPushTime;
 
     private long createTime;
@@ -68,11 +59,9 @@ public DownStreamMsgContext(Message msgExt, Session session, MQConsumerWrapper c
         this.seq = String.valueOf(ServerGlobal.getInstance().getMsgCounter().incrementAndGet());
         this.msgExt = msgExt;
         this.session = session;
-        this.retryTimes = 0;
         this.consumer = consumer;
         this.consumeConcurrentlyContext = consumeConcurrentlyContext;
         this.lastPushTime = System.currentTimeMillis();
-        this.executeTime = System.currentTimeMillis();
         this.createTime = System.currentTimeMillis();
         this.subscriptionItem = subscriptionItem;
         String ttlStr = msgExt.getUserProperties("TTL");
@@ -99,10 +88,6 @@ public void ackMsg() {
         }
     }
 
-    public void delay(long delay) {
-        this.executeTime = System.currentTimeMillis() + (retryTimes + 1) * delay;
-    }
-
     @Override
     public String toString() {
         return "DownStreamMsgContext{" +
@@ -119,19 +104,71 @@ public String toString() {
     }
 
     @Override
-    public int compareTo(Delayed delayed) {
-        DownStreamMsgContext context = (DownStreamMsgContext) delayed;
-        if (this.executeTime > context.executeTime) {
-            return 1;
-        } else if (this.executeTime == context.executeTime) {
-            return 0;
-        } else {
-            return -1;
+    public void retry() {
+        try {
+            logger.info("retry downStream msg start,seq:{},retryTimes:{},bizSeq:{}", this.seq, this.retryTimes, EventMeshUtil.getMessageBizSeq(this.msgExt));
+
+            if (isRetryMsgTimeout(this)) {
+                return;
+            }
+            this.retryTimes++;
+            this.lastPushTime = System.currentTimeMillis();
+
+            Session rechoosen = null;
+            String topic = this.msgExt.getSystemProperties(Constants.PROPERTY_MESSAGE_DESTINATION);
+            if (!SubscriptionMode.BROADCASTING.equals(this.subscriptionItem.getMode())) {
+                rechoosen = this.session.getClientGroupWrapper()
+                        .get().getDownstreamDispatchStrategy().select(this.session.getClientGroupWrapper().get().getSysId()
+                                , topic
+                                , this.session.getClientGroupWrapper().get().getGroupConsumerSessions());
+            } else {
+                rechoosen = this.session;
+            }
+
+            if (rechoosen == null) {
+                logger.warn("retry, found no session to downstream msg,seq:{}, retryTimes:{}, bizSeq:{}", this.seq, this.retryTimes, EventMeshUtil.getMessageBizSeq(this.msgExt));
+            } else {
+                this.session = rechoosen;
+                rechoosen.downstreamMsg(this);
+                logger.info("retry downStream msg end,seq:{},retryTimes:{},bizSeq:{}", this.seq, this.retryTimes, EventMeshUtil.getMessageBizSeq(this.msgExt));
+            }
+        } catch (Exception e) {
+            logger.error("retry-dispatcher error!", e);
         }
     }
 
-    @Override
-    public long getDelay(TimeUnit unit) {
-        return unit.convert(this.executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+    private boolean isRetryMsgTimeout(DownStreamMsgContext downStreamMsgContext) {
+        boolean flag = false;
+        String ttlStr = downStreamMsgContext.msgExt.getUserProperties(EventMeshConstants.PROPERTY_MESSAGE_TTL);
+        long ttl = StringUtils.isNumeric(ttlStr)? Long.parseLong(ttlStr) : EventMeshConstants.DEFAULT_TIMEOUT_IN_MILLISECONDS;;
+
+        String storeTimeStr = downStreamMsgContext.msgExt.getUserProperties(EventMeshConstants.STORE_TIME);
+        long storeTimestamp = StringUtils.isNumeric(storeTimeStr)? Long.parseLong(storeTimeStr) : 0;
+        String leaveTimeStr = downStreamMsgContext.msgExt.getUserProperties(EventMeshConstants.LEAVE_TIME);
+        long brokerCost = StringUtils.isNumeric(leaveTimeStr) ? Long.parseLong(leaveTimeStr) - storeTimestamp : 0;
+
+        String arriveTimeStr = downStreamMsgContext.msgExt.getUserProperties(EventMeshConstants.ARRIVE_TIME);
+        long accessCost = StringUtils.isNumeric(arriveTimeStr) ? System.currentTimeMillis() - Long.parseLong(arriveTimeStr) : 0;
+        double elapseTime = brokerCost + accessCost;
+        if (elapseTime >= ttl) {
+            logger.warn("discard the retry because timeout, seq:{}, retryTimes:{}, bizSeq:{}", downStreamMsgContext.seq, downStreamMsgContext.retryTimes, EventMeshUtil.getMessageBizSeq(downStreamMsgContext.msgExt));
+            flag = true;
+            eventMeshAckMsg(downStreamMsgContext);
+        }
+        return flag;
+    }
+
+    /**
+     * eventMesh ack msg
+     *
+     * @param downStreamMsgContext
+     */
+    private void eventMeshAckMsg(DownStreamMsgContext downStreamMsgContext) {
+        List<Message> msgExts = new ArrayList<Message>();
+        msgExts.add(downStreamMsgContext.msgExt);
+        logger.warn("eventMeshAckMsg topic:{}, seq:{}, bizSeq:{}", downStreamMsgContext.msgExt.getSystemProperties(Constants.PROPERTY_MESSAGE_DESTINATION),
+                downStreamMsgContext.seq, downStreamMsgContext.msgExt.getSystemProperties(EventMeshConstants.PROPERTY_MESSAGE_KEYS));
+        downStreamMsgContext.consumer.updateOffset(msgExts, downStreamMsgContext.consumeConcurrentlyContext);
     }
+
 }