improve topic not found processing

channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/MessageExchangeException.java

@@ -1,8 +1,5 @@
 package com.alibaba.arthas.channel.server.message;
 
-/**
- * @author gongdewei 2020/8/13
- */
 public class MessageExchangeException extends Exception {
     public MessageExchangeException(String message) {
         super(message);

channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/MessageExchangeService.java

@@ -4,6 +4,7 @@ import com.alibaba.arthas.channel.server.message.topic.Topic;
 import reactor.core.publisher.Mono;
 
 /**
+ * Message exchange service for channel-client and channel-server
  * @author gongdewei 2020/8/10
  */
 public interface MessageExchangeService {

channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/TopicNotFoundException.java

@@ -0,0 +1,12 @@
+package com.alibaba.arthas.channel.server.message;
+
+/**
+ * Topic not found exception
+ */
+public class TopicNotFoundException extends MessageExchangeException {
+
+    public TopicNotFoundException(String message) {
+        super(message);
+    }
+
+}

channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/message/impl/MessageExchangeServiceImpl.java

@@ -3,6 +3,7 @@ package com.alibaba.arthas.channel.server.message.impl;
 import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig;
 import com.alibaba.arthas.channel.server.message.MessageExchangeException;
 import com.alibaba.arthas.channel.server.message.MessageExchangeService;
+import com.alibaba.arthas.channel.server.message.TopicNotFoundException;
 import com.alibaba.arthas.channel.server.message.topic.ActionRequestTopic;
 import com.alibaba.arthas.channel.server.message.topic.Topic;
 import org.apache.commons.lang3.StringUtils;
@@ -117,7 +118,7 @@ public class MessageExchangeServiceImpl implements MessageExchangeService {
     private TopicData getAndCheckTopicExists(Topic topic) throws MessageExchangeException {
         TopicData topicData = topicMap.get(topic);
         if (topicData == null) {
-            throw new MessageExchangeException("topic is not exists: " + topic);
+            throw new TopicNotFoundException("topic is not exists: " + topic);
         }
         return topicData;
     }
@@ -128,7 +129,11 @@ public class MessageExchangeServiceImpl implements MessageExchangeService {
             TopicData topicData = getAndCheckTopicExists(topic);
             return Mono.justOrEmpty(topicData.messageQueue.poll(timeout, TimeUnit.MILLISECONDS));
         } catch (Throwable e) {
-            return Mono.error(new MessageExchangeException("poll message failure: "+e.getMessage(), e));
+            if (e instanceof TopicNotFoundException) {
+                return Mono.error(e);
+            } else {
+                return Mono.error(new MessageExchangeException("poll message failure: " + e.getMessage(), e));
+            }
         }
     }
 

channel/channel-server/src/main/java/com/alibaba/arthas/channel/server/web/LegacyApiController.java

@@ -82,7 +82,12 @@ public class LegacyApiController {
                 case INIT_SESSION:
                     actionResponseMono = apiActionDelegateService.initSession(agentId);
                     return convertToApiResponse(actionResponseMono);
+                case INTERRUPT_JOB:
+                    checkSessionId(request);
+                    actionResponseMono = apiActionDelegateService.interruptJob(agentId, request.getSessionId());
+                    return convertToApiResponse(actionResponseMono);
                 case CLOSE_SESSION:
+                    checkSessionId(request);
                     actionResponseMono = apiActionDelegateService.closeSession(agentId, request.getSessionId());
                     return convertToApiResponse(actionResponseMono);
                 default:
@@ -100,6 +105,12 @@ public class LegacyApiController {
         }
     }
 
+    private void checkSessionId(ApiRequest request) {
+        if (StringUtils.isBlank(request.getSessionId())) {
+            throw new IllegalArgumentException("Invalid request, the 'sessionId' is required");
+        }
+    }
+
     private void checkRequestId(ApiRequest request) {
         if (StringUtils.isBlank(request.getRequestId())) {
             throw new IllegalArgumentException("Invalid request, the 'requestId' is required");