[ISSUE #457] message ack problem in retry or repush by closeSession (#…

…459) * modify:optimize flow control in downstreaming msg * modify:optimize stategy of selecting session in downstream msg * modify:optimize msg downstream,msg store in session * modify:fix bug:not a @sharable handler * modify:downstream broadcast msg asynchronously * modify:remove unneccessary interface in eventmesh-connector-api * modify:fix conflict * modify:add license in EventMeshAction * modify:fix ack problem close #457
apache · Jul 28, 2021 · 21cca7c · 21cca7c
1 parent 634262c
commit 21cca7c
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 1 deletion.
diff --git a/...rg/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java b/...rg/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
@@ -305,6 +305,7 @@ private void handleUnackMsgsInSession(Session session) {
                         , session.getClientGroupWrapper().get().groupConsumerSessions);
                 if(reChooseSession != null){
                     downStreamMsgContext.session = reChooseSession;
+                    reChooseSession.getPusher().unAckMsg(downStreamMsgContext.seq, downStreamMsgContext);
                     reChooseSession.downstreamMsg(downStreamMsgContext);
                     logger.info("rePush msg form unAckMsgs,seq:{},rePushClient:{}", entry.getKey(), downStreamMsgContext.session.getClient());
                 }else{

diff --git a/...he/eventmesh/runtime/core/protocol/tcp/client/session/push/retry/EventMeshTcpRetryer.java b/...he/eventmesh/runtime/core/protocol/tcp/client/session/push/retry/EventMeshTcpRetryer.java
@@ -146,6 +146,7 @@ private void retryHandle(DownStreamMsgContext downStreamMsgContext) {
 //                }
             } else {
                 downStreamMsgContext.session = rechoosen;
+                rechoosen.getPusher().unAckMsg(downStreamMsgContext.seq, downStreamMsgContext);
                 rechoosen.downstreamMsg(downStreamMsgContext);
                 logger.info("retry downStream msg end,seq:{},retryTimes:{},bizSeq:{}", downStreamMsgContext.seq, downStreamMsgContext.retryTimes, EventMeshUtil.getMessageBizSeq(downStreamMsgContext.msgExt));
             }

diff --git a/.../main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/MessageAckTask.java b/.../main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/MessageAckTask.java
@@ -50,7 +50,9 @@ public void run() {
             downStreamMsgContext.ackMsg();
             session.getPusher().getUnAckMsg().remove(seq);
         }else {
-           logger.warn("MessageAckTask, seq:{}, downStreamMsgContext not in downStreamMap,client:{}", seq, session.getClient());
+            if(!cmd.equals(Command.RESPONSE_TO_CLIENT_ACK)) {
+                logger.warn("MessageAckTask, seq:{}, downStreamMsgContext not in downStreamMap,client:{}", seq, session.getClient());
+            }
         }
         messageLogger.info("pkg|c2eventMesh|cmd={}|seq=[{}]|user={}|wait={}ms|cost={}ms", cmd, seq, session.getClient(),
                 taskExecuteTime - startTime, System.currentTimeMillis() - startTime);