[ISSUE #711] Consumers can skip the accumulation of messages. (#712)

Co-authored-by: zhangjidi2016 <zhangjidi@cmss.chinamobile.com>

src/main/java/org/apache/rocketmq/console/controller/ConsumerController.java

@@ -62,6 +62,13 @@ public class ConsumerController {
         return consumerService.resetOffset(resetOffsetRequest);
     }
 
+    @RequestMapping(value = "/skipAccumulate.do", method = {RequestMethod.POST})
+    @ResponseBody
+    public Object skipAccumulate(@RequestBody ResetOffsetRequest resetOffsetRequest) {
+        logger.info("op=look resetOffsetRequest={}", JsonUtil.obj2String(resetOffsetRequest));
+        return consumerService.resetOffset(resetOffsetRequest);
+    }
+
     @RequestMapping(value = "/examineSubscriptionGroupConfig.query")
     @ResponseBody
     public Object examineSubscriptionGroupConfig(@RequestParam String consumerGroup) {

src/main/resources/static/src/i18n/zh.js

@@ -31,6 +31,7 @@ var zh = {
     "CONFIG": "配置",
     "SEND_MSG": "发送消息",
     "RESET_CUS_OFFSET": "重置消费位点",
+    "SKIP_MESSAGE_ACCUMULATE":"跳过堆积",
     "DELETE": "删除",
     "CHANGE_LANG": "更换语言",
     "BROKER": "Broker",

src/main/resources/static/src/topic.js

@@ -38,20 +38,20 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http','Notificatio
     $scope.topicShowList = [];
 
     $scope.refreshTopicList = function () {
-            $http({
-                method: "GET",
-                url: "topic/list.query"
-            }).success(function (resp) {
-                if(resp.status ==0){
-                    $scope.allTopicList = resp.data.topicList.sort();
-                    console.log($scope.allTopicList);
-                    console.log(JSON.stringify(resp));
-                    $scope.showTopicList(1,$scope.allTopicList.length);
+        $http({
+            method: "GET",
+            url: "topic/list.query"
+        }).success(function (resp) {
+            if(resp.status ==0){
+                $scope.allTopicList = resp.data.topicList.sort();
+                console.log($scope.allTopicList);
+                console.log(JSON.stringify(resp));
+                $scope.showTopicList(1,$scope.allTopicList.length);
 
-                }else {
-                    Notification.error({message: resp.errMsg, delay: 5000});
-                }
-            });
+            }else {
+                Notification.error({message: resp.errMsg, delay: 5000});
+            }
+        });
     };
 
     $scope.refreshTopicList();
@@ -91,26 +91,26 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http','Notificatio
     };
 
     $scope.filterByType = function(str){
-            if($scope.filterRetry){
-                if(str.startsWith("%R")){
-                    return true
-                }
+        if($scope.filterRetry){
+            if(str.startsWith("%R")){
+                return true
             }
-            if($scope.filterDLQ){
-                if(str.startsWith("%D")){
-                    return true
-                }
+        }
+        if($scope.filterDLQ){
+            if(str.startsWith("%D")){
+                return true
             }
-            if($scope.filterSystem){
-                if(str.startsWith("%S")){
-                    return true
-                }
+        }
+        if($scope.filterSystem){
+            if(str.startsWith("%S")){
+                return true
             }
-            if($scope.filterNormal){
-                if(str.startsWith("%") == false){
-                    return true
-                }
+        }
+        if($scope.filterNormal){
+            if(str.startsWith("%") == false){
+                return true
             }
+        }
         return false;
     };
 
@@ -254,6 +254,33 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http','Notificatio
 
     };
 
+    $scope.openSkipMessageAccumulateDialog = function (topic) {
+        $http({
+            method: "GET",
+            url: "topic/queryTopicConsumerInfo.query",
+            params:{
+                topic:topic
+            }
+        }).success(function (resp) {
+            if(resp.status ==0){
+                if(resp.data.groupList == null){
+                    Notification.error({message: "don't have consume group!", delay: 2000});
+                    return
+                }
+                ngDialog.open({
+                    template: 'skipMessageAccumulateDialog',
+                    controller: 'skipMessageAccumulateDialogController',
+                    data:{
+                        topic: topic,
+                        selectedConsumerGroup:[],
+                        allConsumerGroupList:resp.data.groupList
+                    }
+                });
+            }else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+        });
+    };
 
     $scope.openSendTopicMessageDialog = function (topic) {
         ngDialog.open({
@@ -377,6 +404,34 @@ module.controller('consumerResetOffsetDialogController',['$scope', 'ngDialog', '
     }]
 );
 
+module.controller('skipMessageAccumulateDialogController',['$scope', 'ngDialog', '$http','Notification', function ($scope, ngDialog, $http,Notification) {
+        $scope.skipAccumulate = function () {
+            console.log($scope.ngDialogData.selectedConsumerGroup);
+            $http({
+                method: "POST",
+                url: "consumer/skipAccumulate.do",
+                data: {
+                    resetTime: -1,
+                    consumerGroupList: $scope.ngDialogData.selectedConsumerGroup,
+                    topic:$scope.ngDialogData.topic,
+                    force:true
+                }
+            }).success(function (resp) {
+                if(resp.status ==0){
+                    ngDialog.open({
+                        template: 'resetOffsetResultDialog',
+                        data:{
+                            result:resp.data
+                        }
+                    });
+                }else {
+                    Notification.error({message: resp.errMsg, delay: 2000});
+                }
+            })
+        }
+    }]
+);
+
 module.controller('sendTopicMessageDialogController', ['$scope', 'ngDialog', '$http','Notification',function ($scope, ngDialog, $http,Notification) {
         $scope.sendTopicMessage = {
             topic: $scope.ngDialogData.topic,

src/main/resources/static/view/pages/topic.html

@@ -71,6 +71,10 @@
                                     ng-show="{{!sysFlag}}"
                                     ng-click="openConsumerResetOffsetDialog(topic)">{{'RESET_CUS_OFFSET' | translate}}
                             </button>
+                            <button class="btn btn-raised btn-sm btn-danger" type="button"
+                                    ng-show="{{!sysFlag}}"
+                                    ng-click="openSkipMessageAccumulateDialog(topic)">{{'SKIP_MESSAGE_ACCUMULATE' | translate}}
+                            </button>
                             <button class="btn btn-raised btn-sm btn-danger" type="button"
                                     ng-show="{{!sysFlag}}"
                                     ng-confirm-click="Are you sure to delete?"
@@ -329,6 +333,37 @@
     </div>
 </script>
 
+<script type="text/ng-template" id="skipMessageAccumulateDialog">
+    <div class="modal-header">
+        <h4 class="modal-title">{{topic}} {{'SKIP_MESSAGE_ACCUMULATE'|translate}}</h4>
+    </div>
+    <div class="modal-body">
+        <form class="form-horizontal" novalidate>
+            <div class="form-group">
+                <label class="control-label col-sm-2"> {{ 'SUBSCRIPTION_GROUP' | translate }}:</label>
+                <div class="col-sm-10">
+                    <select name="mySelect" multiple chosen
+                            ng-model="ngDialogData.selectedConsumerGroup"
+                            ng-options="item for item in ngDialogData.allConsumerGroupList"
+                            required>
+                        <option value=""></option>
+                    </select>
+                </div>
+            </div>
+        </form>
+    </div>
+    <div class="modal-footer">
+        <div class="ngdialog-buttons">
+            <button type="button" class="ngdialog-button ngdialog-button-primary"
+                    ng-click="skipAccumulate()">{{ 'COMMIT' | translate }}
+            </button>
+            <button type="button" class="ngdialog-button ngdialog-button-secondary"
+                    ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
+            </button>
+        </div>
+    </div>
+</script>
+
 <script type="text/ng-template" id="statsViewDialog">
     <div class="modal-header">
         <h4 class="modal-title"> [{{ngDialogData.topic}}]{{'STATUS' | translate}}</h4>