Compare commits

..

1 Commits

Author SHA1 Message Date
dependabot[bot]
7b6aaa8978 Bump decode-uri-component from 0.2.0 to 0.2.2 in /frontend
Bumps [decode-uri-component](https://github.com/SamVerschueren/decode-uri-component) from 0.2.0 to 0.2.2.
- [Release notes](https://github.com/SamVerschueren/decode-uri-component/releases)
- [Commits](https://github.com/SamVerschueren/decode-uri-component/compare/v0.2.0...v0.2.2)

---
updated-dependencies:
- dependency-name: decode-uri-component
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
2022-12-07 21:16:45 +00:00
57 changed files with 474 additions and 1091 deletions

1
.gitignore vendored
View File

@@ -5,4 +5,3 @@
.project
.factorypath
.settings/
.vscode

View File

@@ -3577,6 +3577,11 @@
}
}
},
"base16": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/base16/-/base16-1.0.0.tgz",
"integrity": "sha512-pNdYkNPiJUnEhnfXV56+sQy8+AaPcG3POZAUnwr4EeqCUZFz4u2PePbo3e5Gj4ziYPCWGUZT9RHisvJKnwFuBQ=="
},
"base64-js": {
"version": "1.5.1",
"resolved": "https://registry.npmjs.org/base64-js/-/base64-js-1.5.1.tgz",
@@ -4474,6 +4479,14 @@
"sha.js": "2.4.11"
}
},
"cross-fetch": {
"version": "3.1.5",
"resolved": "https://registry.npmjs.org/cross-fetch/-/cross-fetch-3.1.5.tgz",
"integrity": "sha512-lvb1SBsI0Z7GDwmuid+mU3kWVBwTVUbe7S0H52yaaAdQOXq2YktTCZdlAcNKFzE6QtRz0snpw9bNiPeOIkkQvw==",
"requires": {
"node-fetch": "2.6.7"
}
},
"cross-spawn": {
"version": "6.0.5",
"resolved": "https://registry.npmjs.org/cross-spawn/-/cross-spawn-6.0.5.tgz",
@@ -4847,7 +4860,7 @@
"decode-uri-component": {
"version": "0.2.0",
"resolved": "https://registry.npmjs.org/decode-uri-component/-/decode-uri-component-0.2.0.tgz",
"integrity": "sha1-6zkTMzRYd1y4TNGh+uBiEGu4dUU="
"integrity": "sha512-hjf+xovcEn31w/EUYdTXQh/8smFL/dzYjohQGEIgjyNavaJfBY2p5F527Bo1VPATxv0VYTUC2bOcXvqFwk78Og=="
},
"dedent": {
"version": "0.7.0",
@@ -6438,6 +6451,43 @@
"bser": "2.1.1"
}
},
"fbemitter": {
"version": "3.0.0",
"resolved": "https://registry.npmjs.org/fbemitter/-/fbemitter-3.0.0.tgz",
"integrity": "sha512-KWKaceCwKQU0+HPoop6gn4eOHk50bBv/VxjJtGMfwmJt3D29JpN4H4eisCtIPA+a8GVBam+ldMMpMjJUvpDyHw==",
"requires": {
"fbjs": "^3.0.0"
}
},
"fbjs": {
"version": "3.0.4",
"resolved": "https://registry.npmjs.org/fbjs/-/fbjs-3.0.4.tgz",
"integrity": "sha512-ucV0tDODnGV3JCnnkmoszb5lf4bNpzjv80K41wd4k798Etq+UYD0y0TIfalLjZoKgjive6/adkRnszwapiDgBQ==",
"requires": {
"cross-fetch": "^3.1.5",
"fbjs-css-vars": "^1.0.0",
"loose-envify": "^1.0.0",
"object-assign": "^4.1.0",
"promise": "^7.1.1",
"setimmediate": "^1.0.5",
"ua-parser-js": "^0.7.30"
},
"dependencies": {
"promise": {
"version": "7.3.1",
"resolved": "https://registry.npmjs.org/promise/-/promise-7.3.1.tgz",
"integrity": "sha512-nolQXZ/4L+bP/UGlkfaIujX9BKxGwmQ9OT4mOt5yvy8iK1h3wqTEJCijzGANTCCl9nWjY41juyAn2K3Q1hLLTg==",
"requires": {
"asap": "~2.0.3"
}
}
}
},
"fbjs-css-vars": {
"version": "1.0.2",
"resolved": "https://registry.npmjs.org/fbjs-css-vars/-/fbjs-css-vars-1.0.2.tgz",
"integrity": "sha512-b2XGFAFdWZWg0phtAWLHCk836A1Xann+I+Dgd3Gk64MHKZO44FfoD1KxyvbSh0qZsIoXQGGlVztIY+oitJPpRQ=="
},
"figgy-pudding": {
"version": "3.5.2",
"resolved": "https://registry.npmjs.org/figgy-pudding/-/figgy-pudding-3.5.2.tgz",
@@ -6567,6 +6617,15 @@
"readable-stream": "2.3.7"
}
},
"flux": {
"version": "4.0.3",
"resolved": "https://registry.npmjs.org/flux/-/flux-4.0.3.tgz",
"integrity": "sha512-yKAbrp7JhZhj6uiT1FTuVMlIAT1J4jqEyBpFApi1kxpGZCvacMVc/t1pMQyotqHhAgvoE3bNvAykhCo2CLjnYw==",
"requires": {
"fbemitter": "^3.0.0",
"fbjs": "^3.0.1"
}
},
"follow-redirects": {
"version": "1.14.2",
"resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.14.2.tgz",
@@ -9787,11 +9846,21 @@
"resolved": "https://registry.npmjs.org/lodash.clonedeep/-/lodash.clonedeep-4.5.0.tgz",
"integrity": "sha1-4j8/nE+Pvd6HJSnBBxhXoIblzO8="
},
"lodash.curry": {
"version": "4.1.1",
"resolved": "https://registry.npmjs.org/lodash.curry/-/lodash.curry-4.1.1.tgz",
"integrity": "sha512-/u14pXGviLaweY5JI0IUzgzF2J6Ne8INyzAZjImcryjgkZ+ebruBxy2/JaOOkTqScddcYtakjhSaeemV8lR0tA=="
},
"lodash.debounce": {
"version": "4.0.8",
"resolved": "https://registry.npmjs.org/lodash.debounce/-/lodash.debounce-4.0.8.tgz",
"integrity": "sha1-gteb/zCmfEAF/9XiUVMArZyk168="
},
"lodash.flow": {
"version": "3.5.0",
"resolved": "https://registry.npmjs.org/lodash.flow/-/lodash.flow-3.5.0.tgz",
"integrity": "sha512-ff3BX/tSioo+XojX4MOsOMhJw0nZoUEF011LX8g8d3gvjVbxd89cCio4BCXronjxcTUIJUoqKEUA+n4CqvvRPw=="
},
"lodash.memoize": {
"version": "4.1.2",
"resolved": "https://registry.npmjs.org/lodash.memoize/-/lodash.memoize-4.1.2.tgz",
@@ -10291,6 +10360,35 @@
}
}
},
"node-fetch": {
"version": "2.6.7",
"resolved": "https://registry.npmjs.org/node-fetch/-/node-fetch-2.6.7.tgz",
"integrity": "sha512-ZjMPFEfVx5j+y2yF35Kzx5sF7kDzxuDj6ziH4FFbOp87zKDZNx8yExJIb05OGF4Nlt9IHFIMBkRl41VdvcNdbQ==",
"requires": {
"whatwg-url": "^5.0.0"
},
"dependencies": {
"tr46": {
"version": "0.0.3",
"resolved": "https://registry.npmjs.org/tr46/-/tr46-0.0.3.tgz",
"integrity": "sha512-N3WMsuqV66lT30CrXNbEjx4GEwlow3v6rr4mCcv6prnfwhS01rkgyFdjPNBYd9br7LpXV1+Emh01fHnq2Gdgrw=="
},
"webidl-conversions": {
"version": "3.0.1",
"resolved": "https://registry.npmjs.org/webidl-conversions/-/webidl-conversions-3.0.1.tgz",
"integrity": "sha512-2JAn3z8AR6rjK8Sm8orRC0h/bcl/DqL7tRPdGZ4I1CjdF+EaMLmYxBHyXuKL849eucPFhvBoxMsflfOb8kxaeQ=="
},
"whatwg-url": {
"version": "5.0.0",
"resolved": "https://registry.npmjs.org/whatwg-url/-/whatwg-url-5.0.0.tgz",
"integrity": "sha512-saE57nupxk6v3HY35+jzBwYa0rKSy0XR8JSxZPwgLr7ys0IBzhGviA1/TUGJLmSVqs8pb9AnvICXEuOHLprYTw==",
"requires": {
"tr46": "~0.0.3",
"webidl-conversions": "^3.0.0"
}
}
}
},
"node-forge": {
"version": "0.10.0",
"resolved": "https://registry.npmjs.org/node-forge/-/node-forge-0.10.0.tgz",
@@ -12209,6 +12307,11 @@
"resolved": "https://registry.npmjs.org/punycode/-/punycode-2.1.1.tgz",
"integrity": "sha512-XRsRjdf+j5ml+y/6GKHPZbrF/8p2Yga0JPtdqTIY2Xe5ohJPD9saDJJLPvp9+NSBprVvevdXZybnj2cv8OEd0A=="
},
"pure-color": {
"version": "1.3.0",
"resolved": "https://registry.npmjs.org/pure-color/-/pure-color-1.3.0.tgz",
"integrity": "sha512-QFADYnsVoBMw1srW7OVKEYjG+MbIa49s54w1MA1EDY6r2r/sTcKKYqRX1f4GYvnXP7eN/Pe9HFcX+hwzmrXRHA=="
},
"q": {
"version": "1.5.1",
"resolved": "https://registry.npmjs.org/q/-/q-1.5.1.tgz",
@@ -12318,6 +12421,17 @@
"whatwg-fetch": "3.6.2"
}
},
"react-base16-styling": {
"version": "0.6.0",
"resolved": "https://registry.npmjs.org/react-base16-styling/-/react-base16-styling-0.6.0.tgz",
"integrity": "sha512-yvh/7CArceR/jNATXOKDlvTnPKPmGZz7zsenQ3jUwLzHkNUR0CvY3yGYJbWJ/nnxsL8Sgmt5cO3/SILVuPO6TQ==",
"requires": {
"base16": "^1.0.0",
"lodash.curry": "^4.0.1",
"lodash.flow": "^3.3.0",
"pure-color": "^1.2.0"
}
},
"react-dev-utils": {
"version": "11.0.4",
"resolved": "https://registry.npmjs.org/react-dev-utils/-/react-dev-utils-11.0.4.tgz",
@@ -12535,6 +12649,22 @@
"resolved": "https://registry.npmjs.org/react-is/-/react-is-17.0.2.tgz",
"integrity": "sha512-w2GsyukL62IJnlaff/nRegPQR94C/XXamvMWmSHRJ4y7Ts/4ocGRmTHvOs8PSE6pB3dWOrD/nueuU5sduBsQ4w=="
},
"react-json-view": {
"version": "1.21.3",
"resolved": "https://registry.npmjs.org/react-json-view/-/react-json-view-1.21.3.tgz",
"integrity": "sha512-13p8IREj9/x/Ye4WI/JpjhoIwuzEgUAtgJZNBJckfzJt1qyh24BdTm6UQNGnyTq9dapQdrqvquZTo3dz1X6Cjw==",
"requires": {
"flux": "^4.0.1",
"react-base16-styling": "^0.6.0",
"react-lifecycles-compat": "^3.0.4",
"react-textarea-autosize": "^8.3.2"
}
},
"react-lifecycles-compat": {
"version": "3.0.4",
"resolved": "https://registry.npmjs.org/react-lifecycles-compat/-/react-lifecycles-compat-3.0.4.tgz",
"integrity": "sha512-fBASbA6LnOU9dOU2eW7aQ8xmYBSXUIWr+UmF9b1efZBazGNO+rcXT/icdKnYm2pTwcRylVUYwW7H1PHfLekVzA=="
},
"react-refresh": {
"version": "0.8.3",
"resolved": "https://registry.npmjs.org/react-refresh/-/react-refresh-0.8.3.tgz",
@@ -12606,6 +12736,16 @@
"workbox-webpack-plugin": "5.1.4"
}
},
"react-textarea-autosize": {
"version": "8.4.0",
"resolved": "https://registry.npmjs.org/react-textarea-autosize/-/react-textarea-autosize-8.4.0.tgz",
"integrity": "sha512-YrTFaEHLgJsi8sJVYHBzYn+mkP3prGkmP2DKb/tm0t7CLJY5t1Rxix8070LAKb0wby7bl/lf2EeHkuMihMZMwQ==",
"requires": {
"@babel/runtime": "^7.10.2",
"use-composed-ref": "^1.3.0",
"use-latest": "^1.2.1"
}
},
"read-pkg": {
"version": "3.0.0",
"resolved": "https://registry.npmjs.org/read-pkg/-/read-pkg-3.0.0.tgz",
@@ -14709,6 +14849,11 @@
"is-typedarray": "1.0.0"
}
},
"ua-parser-js": {
"version": "0.7.32",
"resolved": "https://registry.npmjs.org/ua-parser-js/-/ua-parser-js-0.7.32.tgz",
"integrity": "sha512-f9BESNVhzlhEFf2CHMSj40NWOjYPl1YKYbrvIr/hFTDEmLq7SRbWvm7FcdcpCYT95zrOhC7gZSxjdnnTpBcwVw=="
},
"unbox-primitive": {
"version": "1.0.1",
"resolved": "https://registry.npmjs.org/unbox-primitive/-/unbox-primitive-1.0.1.tgz",
@@ -14915,6 +15060,24 @@
"resolved": "https://registry.npmjs.org/use/-/use-3.1.1.tgz",
"integrity": "sha512-cwESVXlO3url9YWlFW/TA9cshCEhtu7IKJ/p5soJ/gGpj7vbvFrAY/eIioQ6Dw23KjZhYgiIo8HOs1nQ2vr/oQ=="
},
"use-composed-ref": {
"version": "1.3.0",
"resolved": "https://registry.npmjs.org/use-composed-ref/-/use-composed-ref-1.3.0.tgz",
"integrity": "sha512-GLMG0Jc/jiKov/3Ulid1wbv3r54K9HlMW29IWcDFPEqFkSO2nS0MuefWgMJpeHQ9YJeXDL3ZUF+P3jdXlZX/cQ=="
},
"use-isomorphic-layout-effect": {
"version": "1.1.2",
"resolved": "https://registry.npmjs.org/use-isomorphic-layout-effect/-/use-isomorphic-layout-effect-1.1.2.tgz",
"integrity": "sha512-49L8yCO3iGT/ZF9QttjwLF/ZD9Iwto5LnH5LmEdk/6cFmXddqi2ulF0edxTwjj+7mqvpVVGQWvbXZdn32wRSHA=="
},
"use-latest": {
"version": "1.2.1",
"resolved": "https://registry.npmjs.org/use-latest/-/use-latest-1.2.1.tgz",
"integrity": "sha512-xA+AVm/Wlg3e2P/JiItTziwS7FK92LWrDB0p+hgXloIMuVCeJJ8v6f0eeHyPZaJrM+usM1FkFfbNCrJGs8A/zw==",
"requires": {
"use-isomorphic-layout-effect": "^1.1.1"
}
},
"util": {
"version": "0.11.1",
"resolved": "https://registry.npmjs.org/util/-/util-0.11.1.tgz",
@@ -15083,6 +15246,7 @@
"version": "2.3.2",
"resolved": "https://registry.npmjs.org/braces/-/braces-2.3.2.tgz",
"integrity": "sha512-aNdbnj9P8PjdXU4ybaWLK2IF3jc/EoDYbC7AazW6to3TRsfXxscC9UXOB5iDiEQrkyIbWp2SLQda4+QAa7nc3w==",
"optional": true,
"requires": {
"arr-flatten": "1.1.0",
"array-unique": "0.3.2",
@@ -15100,6 +15264,7 @@
"version": "2.0.1",
"resolved": "https://registry.npmjs.org/extend-shallow/-/extend-shallow-2.0.1.tgz",
"integrity": "sha1-Ua99YUrZqfYQ6huvu5idaxxWiQ8=",
"optional": true,
"requires": {
"is-extendable": "0.1.1"
}
@@ -15130,6 +15295,7 @@
"version": "4.0.0",
"resolved": "https://registry.npmjs.org/fill-range/-/fill-range-4.0.0.tgz",
"integrity": "sha1-1USBHUKPmOsGpj3EAtJAPDKMOPc=",
"optional": true,
"requires": {
"extend-shallow": "2.0.1",
"is-number": "3.0.0",
@@ -15141,6 +15307,7 @@
"version": "2.0.1",
"resolved": "https://registry.npmjs.org/extend-shallow/-/extend-shallow-2.0.1.tgz",
"integrity": "sha1-Ua99YUrZqfYQ6huvu5idaxxWiQ8=",
"optional": true,
"requires": {
"is-extendable": "0.1.1"
}
@@ -15191,6 +15358,7 @@
"version": "3.0.0",
"resolved": "https://registry.npmjs.org/is-number/-/is-number-3.0.0.tgz",
"integrity": "sha1-JP1iAaR4LPUFYcgQJ2r8fRLXEZU=",
"optional": true,
"requires": {
"kind-of": "3.2.2"
},
@@ -15199,6 +15367,7 @@
"version": "3.2.2",
"resolved": "https://registry.npmjs.org/kind-of/-/kind-of-3.2.2.tgz",
"integrity": "sha1-MeohpzS6ubuw8yRm2JOupR5KPGQ=",
"optional": true,
"requires": {
"is-buffer": "1.1.6"
}
@@ -15209,6 +15378,7 @@
"version": "3.1.10",
"resolved": "https://registry.npmjs.org/micromatch/-/micromatch-3.1.10.tgz",
"integrity": "sha512-MWikgl9n9M3w+bpsY3He8L+w9eF9338xRl8IAO5viDizwSzziFEyUzo2xrrloB64ADbTf8uA8vRqqttDTOmccg==",
"optional": true,
"requires": {
"arr-diff": "4.0.0",
"array-unique": "0.3.2",
@@ -15240,6 +15410,7 @@
"version": "2.1.1",
"resolved": "https://registry.npmjs.org/to-regex-range/-/to-regex-range-2.1.1.tgz",
"integrity": "sha1-fIDBe53+vlmeJzZ+DU3VWQFB2zg=",
"optional": true,
"requires": {
"is-number": "3.0.0",
"repeat-string": "1.6.1"

View File

@@ -3921,9 +3921,9 @@ decimal.js@^10.2.0:
integrity sha512-KaL7+6Fw6i5A2XSnsbhm/6B+NuEA7TZ4vqxnd5tXz9sbKtrN9Srj8ab4vKVdK8YAqZO9P1kg45Y6YLoduPf+kw==
decode-uri-component@^0.2.0:
version "0.2.0"
resolved "https://registry.yarnpkg.com/decode-uri-component/-/decode-uri-component-0.2.0.tgz#eb3913333458775cb84cd1a1fae062106bb87545"
integrity sha1-6zkTMzRYd1y4TNGh+uBiEGu4dUU=
version "0.2.2"
resolved "https://registry.yarnpkg.com/decode-uri-component/-/decode-uri-component-0.2.2.tgz#e69dbe25d37941171dd540e024c444cd5188e1e9"
integrity sha512-FqUYQ+8o158GyGTrMFJms9qh3CqTKvAqgqsTnkLI8sKu0028orqBhxNMFkFen0zGyg6epACD32pjVk58ngIErQ==
dedent@^0.7.0:
version "0.7.0"

View File

@@ -91,10 +91,10 @@
<commons-io.version>2.4</commons-io.version>
<commons-cli.version>1.2</commons-cli.version>
<commons-collections.version>3.2.2</commons-collections.version>
<rocketmq.version>5.1.0</rocketmq.version>
<rocketmq.version>4.9.3</rocketmq.version>
<surefire.version>2.19.1</surefire.version>
<aspectj.version>1.9.6</aspectj.version>
<lombok.version>1.18.22</lombok.version>
<lombok.version>1.18.12</lombok.version>
<main.basedir>${basedir}/../..</main.basedir>
<docker.image.prefix>apacherocketmq</docker.image.prefix>
<spring.boot.version>2.6.0</spring.boot.version>

View File

@@ -21,7 +21,7 @@ import org.apache.commons.collections.MapUtils;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.tools.admin.MQAdminExt;
@Slf4j

View File

@@ -19,7 +19,7 @@ package org.apache.rocketmq.dashboard.controller;
import com.google.common.base.Preconditions;
import javax.annotation.Resource;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.dashboard.model.ConnectionInfo;
import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
import org.apache.rocketmq.dashboard.model.request.DeleteSubGroupRequest;

View File

@@ -18,7 +18,7 @@ package org.apache.rocketmq.dashboard.controller;
import com.google.common.collect.Maps;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.dashboard.model.MessagePage;
import org.apache.rocketmq.dashboard.model.MessageView;
import org.apache.rocketmq.dashboard.model.request.MessageQuery;

View File

@@ -17,7 +17,7 @@
package org.apache.rocketmq.dashboard.controller;
import javax.annotation.Resource;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
import org.apache.rocketmq.common.protocol.body.ProducerConnection;
import org.apache.rocketmq.dashboard.model.ConnectionInfo;
import org.apache.rocketmq.dashboard.permisssion.Permission;
import org.apache.rocketmq.dashboard.service.ProducerService;

View File

@@ -56,12 +56,6 @@ public class TopicController {
return topicService.fetchAllTopicList(skipSysProcess, skipRetryAndDlq);
}
@RequestMapping(value = "/list.queryTopicType", method = RequestMethod.GET)
@ResponseBody
public Object listTopicType() {
return topicService.examineAllTopicType();
}
@RequestMapping(value = "/stats.query", method = RequestMethod.GET)
@ResponseBody
public Object stats(@RequestParam String topic) {

View File

@@ -20,7 +20,7 @@ import com.google.common.collect.Sets;
import java.util.Collection;
import java.util.HashSet;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.common.protocol.body.Connection;
public class ConnectionInfo extends Connection {
private String versionDesc;

View File

@@ -16,7 +16,7 @@
*/
package org.apache.rocketmq.dashboard.model;
import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
import org.apache.rocketmq.common.admin.RollbackStats;
import com.google.common.collect.Lists;
import java.util.List;

View File

@@ -17,8 +17,8 @@
package org.apache.rocketmq.dashboard.model;
import lombok.Data;
import org.apache.rocketmq.remoting.protocol.body.CMResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.CMResult;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
@Data
public class DlqMessageResendResult {

View File

@@ -16,8 +16,8 @@
*/
package org.apache.rocketmq.dashboard.model;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
private String group;
@@ -27,8 +27,6 @@ public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
private MessageModel messageModel;
private int consumeTps;
private long diffTotal = -1;
private String subGroupType = "NORMAL";
public String getGroup() {
return group;
@@ -93,12 +91,4 @@ public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
public void setVersion(String version) {
this.version = version;
}
public String getSubGroupType() {
return subGroupType;
}
public void setSubGroupType(String subGroupType) {
this.subGroupType = subGroupType;
}
}

View File

@@ -16,7 +16,7 @@
*/
package org.apache.rocketmq.dashboard.model;
import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
import org.apache.rocketmq.common.admin.OffsetWrapper;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.BeanUtils;

View File

@@ -16,7 +16,7 @@
*/
package org.apache.rocketmq.dashboard.model.request;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import java.util.List;

View File

@@ -15,6 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.dashboard.model.request;
import com.google.common.base.Objects;
import java.util.List;
@@ -31,7 +32,6 @@ public class TopicConfigInfo {
private int perm;
private boolean order;
private String messageType;
public List<String> getClusterNameList() {
return clusterNameList;
}
@@ -92,18 +92,6 @@ public class TopicConfigInfo {
this.order = order;
}
public String getMessageType() {
return messageType;
}
public void setMessageType(String messageType) {
this.messageType = messageType;
}
@Override
public boolean equals(Object o) {
if (this == o)
@@ -115,13 +103,12 @@ public class TopicConfigInfo {
readQueueNums == that.readQueueNums &&
perm == that.perm &&
order == that.order &&
Objects.equal(topicName, that.topicName) &&
Objects.equal(messageType, that.messageType);
Objects.equal(topicName, that.topicName);
}
@Override
public int hashCode() {
return Objects.hashCode(topicName, writeQueueNums, readQueueNums, perm, order,messageType);
return Objects.hashCode(topicName, writeQueueNums, readQueueNums, perm, order);
}
}

View File

@@ -1,45 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.dashboard.model.request;
import java.util.List;
public class TopicTypeList {
private List<String> topicNameList;
private List<String> messageTypeList;
public List<String> getTopicNameList() {
return topicNameList;
}
public void setTopicNameList(List<String> topicNameList) {
this.topicNameList = topicNameList;
}
public List<String> getMessageTypeList() {
return messageTypeList;
}
public void setMessageTypeList(List<String> messageTypeList) {
this.messageTypeList = messageTypeList;
}
public TopicTypeList(List<String> topicNameList, List<String> messageTypeList) {
this.topicNameList = topicNameList;
this.messageTypeList = messageTypeList;
}
}

View File

@@ -1,38 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.dashboard.model.request;
public class TopicTypeMeta {
private String topicName;
private String messageType;
public String getTopicName() {
return topicName;
}
public void setTopicName(String topicName) {
this.topicName = topicName;
}
public String getMessageType() {
return messageType;
}
public void setMessageType(String messageType) {
this.messageType = messageType;
}
}

View File

@@ -19,8 +19,8 @@ package org.apache.rocketmq.dashboard.service;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import com.google.common.base.Throwables;
import com.google.common.collect.Sets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.annotation.Resource;
import org.apache.commons.collections.CollectionUtils;
@@ -28,7 +28,7 @@ import org.apache.commons.collections.CollectionUtils;
public abstract class AbstractCommonService {
@Resource
protected MQAdminExt mqAdminExt;
protected final Set<String> changeToBrokerNameSet(Map<String, Set<String>> clusterAddrTable,
protected final Set<String> changeToBrokerNameSet(HashMap<String, Set<String>> clusterAddrTable,
List<String> clusterNameList, List<String> brokerNameList) {
Set<String> finalBrokerNameList = Sets.newHashSet();
if (CollectionUtils.isNotEmpty(clusterNameList)) {
@@ -38,8 +38,7 @@ public abstract class AbstractCommonService {
}
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
if (CollectionUtils.isNotEmpty(brokerNameList)) {

View File

@@ -17,8 +17,8 @@
package org.apache.rocketmq.dashboard.service;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.dashboard.model.ConsumerGroupRollBackStat;
import org.apache.rocketmq.dashboard.model.GroupConsumeInfo;
import org.apache.rocketmq.dashboard.model.TopicConsumerInfo;

View File

@@ -19,7 +19,7 @@ package org.apache.rocketmq.dashboard.service;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.dashboard.model.MessagePage;
import org.apache.rocketmq.dashboard.model.request.MessageQuery;
import org.apache.rocketmq.tools.admin.api.MessageTrack;

View File

@@ -17,7 +17,7 @@
package org.apache.rocketmq.dashboard.service;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
import org.apache.rocketmq.common.protocol.body.ProducerConnection;
public interface ProducerService {
ProducerConnection getProducerConnection(String producerGroup, String topic);

View File

@@ -19,11 +19,10 @@ package org.apache.rocketmq.dashboard.service;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.dashboard.model.request.TopicTypeList;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
@@ -32,8 +31,6 @@ import java.util.List;
public interface TopicService {
TopicList fetchAllTopicList(boolean skipSysProcess, boolean skipRetryAndDlq);
TopicTypeList examineAllTopicType();
TopicStatsTable stats(String topic);
TopicRouteData route(String topic);

View File

@@ -29,41 +29,31 @@ import org.apache.rocketmq.client.impl.MQAdminImpl;
import org.apache.rocketmq.common.AclConfig;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.admin.RollbackStats;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.message.MessageRequestMode;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
import org.apache.rocketmq.remoting.protocol.body.ClusterAclVersionInfo;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo;
import org.apache.rocketmq.remoting.protocol.body.QueryConsumeQueueResponseBody;
import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail;
import org.apache.rocketmq.remoting.protocol.subscription.GroupForbidden;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
import org.apache.rocketmq.common.protocol.body.ClusterAclVersionInfo;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.body.ProducerConnection;
import org.apache.rocketmq.common.protocol.body.QueryConsumeQueueResponseBody;
import org.apache.rocketmq.common.protocol.body.QueueTimeSpan;
import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.dashboard.util.JsonUtil;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
@@ -73,9 +63,7 @@ import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.apache.rocketmq.tools.admin.api.BrokerOperatorResult;
import org.apache.rocketmq.tools.admin.api.MessageTrack;
import org.apache.rocketmq.tools.admin.common.AdminToolResult;
import org.joor.Reflect;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -93,7 +81,7 @@ public class MQAdminExtImpl implements MQAdminExt {
@Override
public void updateBrokerConfig(String brokerAddr, Properties properties)
throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
UnsupportedEncodingException, InterruptedException, MQBrokerException, MQClientException {
UnsupportedEncodingException, InterruptedException, MQBrokerException {
MQAdminInstance.threadLocalMQAdminExt().updateBrokerConfig(brokerAddr, properties);
}
@@ -140,7 +128,7 @@ public class MQAdminExtImpl implements MQAdminExt {
}
@Override
public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) throws MQBrokerException {
public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) {
RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, null);
RemotingCommand response = null;
@@ -148,8 +136,7 @@ public class MQAdminExtImpl implements MQAdminExt {
response = remotingClient.invokeSync(addr, request, 3000);
}
catch (Exception err) {
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
assert response != null;
switch (response.getCode()) {
@@ -158,12 +145,12 @@ public class MQAdminExtImpl implements MQAdminExt {
return subscriptionGroupWrapper.getSubscriptionGroupTable().get(group);
}
default:
throw new MQBrokerException(response.getCode(), response.getRemark());
throw Throwables.propagate(new MQBrokerException(response.getCode(), response.getRemark()));
}
}
@Override
public TopicConfig examineTopicConfig(String addr, String topic) throws MQBrokerException {
public TopicConfig examineTopicConfig(String addr, String topic) {
RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
RemotingCommand response = null;
@@ -171,8 +158,7 @@ public class MQAdminExtImpl implements MQAdminExt {
response = remotingClient.invokeSync(addr, request, 3000);
}
catch (Exception err) {
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
@@ -180,7 +166,7 @@ public class MQAdminExtImpl implements MQAdminExt {
return topicConfigSerializeWrapper.getTopicConfigTable().get(topic);
}
default:
throw new MQBrokerException(response.getCode(), response.getRemark());
throw Throwables.propagate(new MQBrokerException(response.getCode(), response.getRemark()));
}
}
@@ -390,14 +376,14 @@ public class MQAdminExtImpl implements MQAdminExt {
}
@Override
public void createTopic(String key, String newTopic, int queueNum, Map<String, String> attributes) throws MQClientException {
MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, attributes);
public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum);
}
@Override
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes)
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
throws MQClientException {
MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, topicSysFlag, attributes);
MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, topicSysFlag);
}
@Override
@@ -586,257 +572,4 @@ public class MQAdminExtImpl implements MQAdminExt {
String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
return false;
}
@Override
public void addBrokerToContainer(String brokerContainerAddr, String brokerConfig) throws InterruptedException,
MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'addBrokerToContainer'");
}
@Override
public void removeBrokerFromContainer(String brokerContainerAddr, String clusterName, String brokerName,
long brokerId) throws InterruptedException, MQBrokerException, RemotingTimeoutException,
RemotingSendRequestException, RemotingConnectException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'removeBrokerFromContainer'");
}
@Override
public void updateGlobalWhiteAddrConfig(String addr, String globalWhiteAddrs, String aclFileFullPath)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'updateGlobalWhiteAddrConfig'");
}
@Override
public TopicStatsTable examineTopicStats(String brokerAddr, String topic)
throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'examineTopicStats'");
}
@Override
public AdminToolResult<TopicStatsTable> examineTopicStatsConcurrent(String topic) {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'examineTopicStatsConcurrent'");
}
@Override
public ConsumeStats examineConsumeStats(String brokerAddr, String consumerGroup, String topicName,
long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
RemotingConnectException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'examineConsumeStats'");
}
@Override
public AdminToolResult<ConsumeStats> examineConsumeStatsConcurrent(String consumerGroup, String topic) {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'examineConsumeStatsConcurrent'");
}
@Override
public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup, String brokerAddr)
throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'examineConsumerConnectionInfo'");
}
@Override
public ProducerTableInfo getAllProducerInfo(String brokerAddr)
throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'getAllProducerInfo'");
}
@Override
public void deleteTopic(String topicName, String clusterName)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'deleteTopic'");
}
@Override
public AdminToolResult<BrokerOperatorResult> deleteTopicInBrokerConcurrent(Set<String> addrs, String topic) {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'deleteTopicInBrokerConcurrent'");
}
@Override
public void deleteTopicInNameServer(Set<String> addrs, String clusterName, String topic)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'deleteTopicInNameServer'");
}
@Override
public AdminToolResult<BrokerOperatorResult> resetOffsetNewConcurrent(String group, String topic, long timestamp) {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'resetOffsetNewConcurrent'");
}
@Override
public TopicList queryTopicsByConsumer(String group)
throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'queryTopicsByConsumer'");
}
@Override
public AdminToolResult<TopicList> queryTopicsByConsumerConcurrent(String group) {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'queryTopicsByConsumerConcurrent'");
}
@Override
public SubscriptionData querySubscription(String group, String topic)
throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'querySubscription'");
}
@Override
public AdminToolResult<List<QueueTimeSpan>> queryConsumeTimeSpanConcurrent(String topic, String group) {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'queryConsumeTimeSpanConcurrent'");
}
@Override
public boolean deleteExpiredCommitLog(String cluster) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, MQClientException, InterruptedException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'deleteExpiredCommitLog'");
}
@Override
public boolean deleteExpiredCommitLogByAddr(String addr) throws RemotingConnectException,
RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'deleteExpiredCommitLogByAddr'");
}
@Override
public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack,
boolean metrics) throws RemotingException, MQClientException, InterruptedException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'getConsumerRunningInfo'");
}
@Override
public List<MessageTrack> messageTrackDetailConcurrent(MessageExt msg)
throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'messageTrackDetailConcurrent'");
}
@Override
public void setMessageRequestMode(String brokerAddr, String topic, String consumerGroup, MessageRequestMode mode,
int popWorkGroupSize, long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
RemotingSendRequestException, RemotingConnectException, MQClientException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'setMessageRequestMode'");
}
@Override
public long searchOffset(String brokerAddr, String topicName, int queueId, long timestamp, long timeoutMillis)
throws RemotingException, MQBrokerException, InterruptedException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'searchOffset'");
}
@Override
public void resetOffsetByQueueId(String brokerAddr, String consumerGroup, String topicName, int queueId,
long resetOffset) throws RemotingException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'resetOffsetByQueueId'");
}
@Override
public void createStaticTopic(String addr, String defaultTopic, TopicConfig topicConfig,
TopicQueueMappingDetail mappingDetail, boolean force)
throws RemotingException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'createStaticTopic'");
}
@Override
public GroupForbidden updateAndGetGroupReadForbidden(String brokerAddr, String groupName, String topicName,
Boolean readable) throws RemotingException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'updateAndGetGroupReadForbidden'");
}
@Override
public MessageExt queryMessage(String clusterName, String topic, String msgId)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'queryMessage'");
}
@Override
public HARuntimeInfo getBrokerHAStatus(String brokerAddr) throws RemotingConnectException,
RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'getBrokerHAStatus'");
}
@Override
public BrokerReplicasInfo getInSyncStateData(String controllerAddress, List<String> brokers)
throws RemotingException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'getInSyncStateData'");
}
@Override
public EpochEntryCache getBrokerEpochCache(String brokerAddr)
throws RemotingException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'getBrokerEpochCache'");
}
@Override
public GetMetaDataResponseHeader getControllerMetaData(String controllerAddr)
throws RemotingException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'getControllerMetaData'");
}
@Override
public void resetMasterFlushOffset(String brokerAddr, long masterFlushOffset) throws InterruptedException,
MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'resetMasterFlushOffset'");
}
@Override
public Map<String, Properties> getControllerConfig(List<String> controllerServers)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
RemotingConnectException, MQClientException, UnsupportedEncodingException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'getControllerConfig'");
}
@Override
public void updateControllerConfig(Properties properties, List<String> controllers)
throws InterruptedException, RemotingConnectException, UnsupportedEncodingException,
RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'updateControllerConfig'");
}
@Override
public ElectMasterResponseHeader electMaster(String controllerAddr, String clusterName, String brokerName,
String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'electMaster'");
}
@Override
public void cleanControllerBrokerData(String controllerAddr, String clusterName, String brokerName,
String brokerAddr, boolean isCleanLivingBroker)
throws RemotingException, InterruptedException, MQBrokerException {
// TODO Auto-generated method stub
throw new UnsupportedOperationException("Unimplemented method 'cleanControllerBrokerData'");
}
}

View File

@@ -36,8 +36,8 @@ import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.AclConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.dashboard.model.request.AclRequest;
import org.apache.rocketmq.dashboard.service.AbstractCommonService;
import org.apache.rocketmq.dashboard.service.AclService;
@@ -68,8 +68,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
}
} catch (Exception e) {
log.error("getAclConfig error.", e);
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
AclConfig aclConfig = new AclConfig();
aclConfig.setGlobalWhiteAddrs(Collections.emptyList());
@@ -101,8 +100,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
mqAdminExt.createAndUpdatePlainAccessConfig(addr, config);
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -118,8 +116,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
log.info("Delete acl [{}] from broker [{}] complete", config.getAccessKey(), addr);
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -145,8 +142,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
mqAdminExt.createAndUpdatePlainAccessConfig(addr, config);
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -178,8 +174,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
}
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -211,8 +206,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
}
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -255,8 +249,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
}
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -268,8 +261,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
mqAdminExt.createAndUpdatePlainAccessConfig(addr, config);
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -289,8 +281,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
mqAdminExt.updateGlobalWhiteAddrConfig(addr, StringUtils.join(aclConfig.getGlobalWhiteAddrs(), ","));
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -306,8 +297,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
mqAdminExt.updateGlobalWhiteAddrConfig(addr, StringUtils.join(aclConfig.getGlobalWhiteAddrs(), ","));
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -321,8 +311,7 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
mqAdminExt.updateGlobalWhiteAddrConfig(addr, StringUtils.join(whiteList, ","));
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}

View File

@@ -17,10 +17,9 @@
package org.apache.rocketmq.dashboard.service.impl;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.apache.rocketmq.dashboard.service.ClusterService;
import org.apache.rocketmq.dashboard.util.JsonUtil;
@@ -31,10 +30,8 @@ import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
import java.util.stream.Collectors;
@Service
public class ClusterServiceImpl implements ClusterService {
@@ -59,14 +56,10 @@ public class ClusterServiceImpl implements ClusterService {
}
resultMap.put("clusterInfo", clusterInfo);
resultMap.put("brokerServer", brokerServer);
// add messageType
resultMap.put("messageTypes", Arrays.stream(TopicMessageType.values()).sorted()
.collect(Collectors.toMap(TopicMessageType::getValue, messageType ->String.format("MESSAGE_TYPE_%s",messageType.getValue()))));
return resultMap;
}
catch (Exception err) {
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
}
@@ -77,8 +70,7 @@ public class ClusterServiceImpl implements ClusterService {
return mqAdminExt.getBrokerConfig(brokerAddr);
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
}

View File

@@ -44,18 +44,18 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.admin.RollbackStats;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.Connection;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.dashboard.config.RMQConfigure;
import org.apache.rocketmq.dashboard.model.ConsumerGroupRollBackStat;
@@ -73,6 +73,8 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;
import static com.google.common.base.Throwables.propagate;
@Service
public class ConsumerServiceImpl extends AbstractCommonService implements ConsumerService, InitializingBean, DisposableBean {
private Logger logger = LoggerFactory.getLogger(ConsumerServiceImpl.class);
@@ -129,8 +131,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
}
catch (Exception err) {
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
List<GroupConsumeInfo> groupConsumeInfoList = Collections.synchronizedList(Lists.newArrayList());
CountDownLatch countDownLatch = new CountDownLatch(consumerGroupSet.size());
@@ -177,10 +178,6 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
ConsumerConnection consumerConnection = null;
boolean isFifoType = examineSubscriptionGroupConfig(consumerGroup)
.stream().map(ConsumerConfigInfo::getSubscriptionGroupConfig)
.allMatch(SubscriptionGroupConfig::isConsumeMessageOrderly);
try {
consumerConnection = mqAdminExt.examineConsumerConnectionInfo(consumerGroup);
}
@@ -189,13 +186,6 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
groupConsumeInfo.setGroup(consumerGroup);
if (SYSTEM_GROUP_SET.contains(consumerGroup)) {
groupConsumeInfo.setSubGroupType("SYSTEM");
} else if (isFifoType) {
groupConsumeInfo.setSubGroupType("FIFO");
} else {
groupConsumeInfo.setSubGroupType("NORMAL");
}
if (consumeStats != null) {
groupConsumeInfo.setConsumeTps((int)consumeStats.getConsumeTps());
@@ -228,8 +218,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
consumeStats = mqAdminExt.examineConsumeStats(groupName, topic);
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw propagate(e);
}
List<MessageQueue> mqList = Lists.newArrayList(Iterables.filter(consumeStats.getOffsetTable().keySet(), new Predicate<MessageQueue>() {
@Override
@@ -289,8 +278,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
return group2ConsumerInfoMap;
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw propagate(e);
}
}
@@ -353,8 +341,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw propagate(e);
}
return consumerConfigInfoList;
}
@@ -379,8 +366,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw propagate(e);
}
return true;
}
@@ -407,8 +393,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
}
catch (Exception err) {
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
return true;
}
@@ -423,8 +408,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
}
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
return brokerNameSet;
@@ -436,8 +420,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
return mqAdminExt.examineConsumerConnectionInfo(consumerGroup);
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -447,8 +430,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
return mqAdminExt.getConsumerRunningInfo(consumerGroup, clientId, jstack);
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
}

View File

@@ -107,8 +107,7 @@ public class DashboardCollectServiceImpl implements DashboardCollectService {
strings = Files.readLines(file, Charsets.UTF_8);
}
catch (IOException e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
StringBuffer sb = new StringBuffer();
for (String string : strings) {

View File

@@ -25,8 +25,8 @@ import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.dashboard.model.DlqMessageResendResult;
import org.apache.rocketmq.dashboard.model.DlqMessageRequest;
import org.apache.rocketmq.dashboard.model.MessagePage;
@@ -62,12 +62,10 @@ public class DlqMessageServiceImpl implements DlqMessageService {
&& e.getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) {
return new MessagePage(new PageImpl<>(messageViews, page, 0), query.getTaskId());
} else {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
return messageService.queryMessageByPage(query);
}

View File

@@ -37,9 +37,9 @@ import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.Connection;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.dashboard.config.RMQConfigure;
import org.apache.rocketmq.dashboard.exception.ServiceException;
import org.apache.rocketmq.dashboard.model.QueueOffsetInfo;
@@ -115,8 +115,7 @@ public class MessageServiceImpl implements MessageService {
if (err instanceof MQClientException) {
throw new ServiceException(-1, ((MQClientException) err).getErrorMessage());
}
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
}
@@ -186,8 +185,7 @@ public class MessageServiceImpl implements MessageService {
});
return messageViewList;
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
} finally {
consumer.shutdown();
}
@@ -211,8 +209,7 @@ public class MessageServiceImpl implements MessageService {
try {
return mqAdminExt.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -226,8 +223,7 @@ public class MessageServiceImpl implements MessageService {
return mqAdminExt.consumeMessageDirectly(consumerGroup, connection.getClientId(), topic, msgId);
}
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
throw new IllegalStateException("NO CONSUMER");
@@ -392,8 +388,7 @@ public class MessageServiceImpl implements MessageService {
PageImpl<MessageView> page = new PageImpl<>(messageViews, query.page(), total);
return new MessagePageTask(page, queueOffsetInfos);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
} finally {
consumer.shutdown();
}
@@ -460,8 +455,7 @@ public class MessageServiceImpl implements MessageService {
}
return new PageImpl<>(messageViews, query.page(), total);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
} finally {
consumer.shutdown();
}

View File

@@ -82,8 +82,7 @@ public class MonitorServiceImpl implements MonitorService {
MixAll.string2File(dataStr, path);
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}

View File

@@ -19,7 +19,7 @@ package org.apache.rocketmq.dashboard.service.impl;
import com.google.common.base.Throwables;
import javax.annotation.Resource;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
import org.apache.rocketmq.common.protocol.body.ProducerConnection;
import org.apache.rocketmq.dashboard.service.ProducerService;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.springframework.stereotype.Service;
@@ -35,8 +35,7 @@ public class ProducerServiceImpl implements ProducerService {
return mqAdminExt.examineProducerConnectionInfo(producerGroup, topic);
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
}

View File

@@ -18,57 +18,43 @@
package org.apache.rocketmq.dashboard.service.impl;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.trace.TraceContext;
import org.apache.rocketmq.client.trace.TraceDispatcher;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.dashboard.config.RMQConfigure;
import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
import org.apache.rocketmq.dashboard.model.request.TopicTypeList;
import org.apache.rocketmq.dashboard.model.request.TopicTypeMeta;
import org.apache.rocketmq.dashboard.service.AbstractCommonService;
import org.apache.rocketmq.dashboard.service.TopicService;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.tools.command.CommandUtil;
import org.joor.Reflect;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import static org.apache.rocketmq.common.TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE;
@Service
public class TopicServiceImpl extends AbstractCommonService implements TopicService {
@@ -82,59 +68,23 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
TopicList allTopics = mqAdminExt.fetchAllTopicList();
TopicList sysTopics = getSystemTopicList();
Set<String> topics =
allTopics.getTopicList().stream().map(topic -> {
if (!skipSysProcess && sysTopics.getTopicList().contains(topic)) {
topic = String.format("%s%s", "%SYS%", topic);
}
return topic;
}).filter(topic -> {
if (skipRetryAndDlq) {
return !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
|| topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX));
}
return true;
}).collect(Collectors.toSet());
allTopics.getTopicList().stream().map(topic -> {
if (!skipSysProcess && sysTopics.getTopicList().contains(topic)) {
topic = String.format("%s%s", "%SYS%", topic);
}
return topic;
}).filter(topic -> {
if (skipRetryAndDlq) {
return !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
|| topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX));
}
return true;
}).collect(Collectors.toSet());
allTopics.getTopicList().clear();
allTopics.getTopicList().addAll(topics);
return allTopics;
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
}
}
@Override
public TopicTypeList examineAllTopicType() {
ArrayList<TopicTypeMeta> topicTypes = new ArrayList<>();
ArrayList<String> names = new ArrayList<>();
ArrayList<String> messageTypes = new ArrayList<>();
TopicList topicList = fetchAllTopicList(false, false);
checkTopicType(topicList, topicTypes);
topicTypes.sort((t1, t2) -> t1.getTopicName().compareTo(t2.getTopicName()));
for (TopicTypeMeta topicTypeMeta : topicTypes) {
names.add(topicTypeMeta.getTopicName());
messageTypes.add(topicTypeMeta.getMessageType());
}
return new TopicTypeList(names, messageTypes);
}
private void checkTopicType(TopicList topicList, ArrayList<TopicTypeMeta> topicTypes) {
for (String topicName : topicList.getTopicList()) {
TopicTypeMeta topicType = new TopicTypeMeta();
topicType.setTopicName(topicName);
if (topicName.startsWith("%R")) {
topicType.setMessageType("RETRY");
} else if (topicName.startsWith("%D")) {
topicType.setMessageType("DELAY");
} else if (topicName.startsWith("%S")) {
topicType.setMessageType("SYSTEM");
} else {
List<TopicConfigInfo> topicConfigInfos = examineTopicConfig(topicName);
if (!CollectionUtils.isEmpty(topicConfigInfos)) {
topicType.setMessageType(topicConfigInfos.get(0).getMessageType());
}
}
topicTypes.add(topicType);
throw Throwables.propagate(e);
}
}
@@ -143,8 +93,7 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
try {
return mqAdminExt.examineTopicStats(topic);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -153,8 +102,7 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
try {
return mqAdminExt.examineTopicRouteInfo(topic);
} catch (Exception ex) {
Throwables.throwIfUnchecked(ex);
throw new RuntimeException(ex);
throw Throwables.propagate(ex);
}
}
@@ -163,8 +111,7 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
try {
return mqAdminExt.queryTopicConsumeByWho(topic);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -172,20 +119,14 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
public void createOrUpdate(TopicConfigInfo topicCreateOrUpdateRequest) {
TopicConfig topicConfig = new TopicConfig();
BeanUtils.copyProperties(topicCreateOrUpdateRequest, topicConfig);
String messageType = topicCreateOrUpdateRequest.getMessageType();
if (StringUtils.isBlank(messageType)) {
messageType = TopicMessageType.NORMAL.name();
}
topicConfig.setAttributes(ImmutableMap.of("+".concat(TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName()), messageType));
try {
ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
for (String brokerName : changeToBrokerNameSet(clusterInfo.getClusterAddrTable(),
topicCreateOrUpdateRequest.getClusterNameList(), topicCreateOrUpdateRequest.getBrokerNameList())) {
topicCreateOrUpdateRequest.getClusterNameList(), topicCreateOrUpdateRequest.getBrokerNameList())) {
mqAdminExt.createAndUpdateTopicConfig(clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr(), topicConfig);
}
} catch (Exception err) {
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
}
@@ -196,8 +137,7 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
clusterInfo = mqAdminExt.examineBrokerClusterInfo();
return mqAdminExt.examineTopicConfig(clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr(), topic);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -210,11 +150,6 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
TopicConfig topicConfig = examineTopicConfig(topic, brokerData.getBrokerName());
BeanUtils.copyProperties(topicConfig, topicConfigInfo);
topicConfigInfo.setBrokerNameList(Lists.newArrayList(brokerData.getBrokerName()));
String messageType = topicConfig.getAttributes().get(TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName());
if (StringUtils.isBlank(messageType)) {
messageType = TopicMessageType.UNSPECIFIED.name();
}
topicConfigInfo.setMessageType(messageType);
topicConfigInfoList.add(topicConfigInfo);
}
return topicConfigInfoList;
@@ -235,8 +170,7 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
}
mqAdminExt.deleteTopicInNameServer(nameServerSet, topic);
} catch (Exception err) {
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
return true;
}
@@ -247,8 +181,7 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
try {
clusterInfo = mqAdminExt.examineBrokerClusterInfo();
} catch (Exception err) {
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
for (String clusterName : clusterInfo.getClusterAddrTable().keySet()) {
deleteTopic(topic, clusterName);
@@ -264,13 +197,11 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
try {
clusterInfo = mqAdminExt.examineBrokerClusterInfo();
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
mqAdminExt.deleteTopicInBroker(Sets.newHashSet(clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr()), topic);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
return true;
}
@@ -285,12 +216,6 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
return defaultMQProducer;
}
public TransactionMQProducer buildTransactionMQProducer(String producerGroup, RPCHook rpcHook, boolean traceEnabled) {
TransactionMQProducer defaultMQProducer = new TransactionMQProducer(null, producerGroup, rpcHook, traceEnabled, TopicValidator.RMQ_SYS_TRACE_TOPIC);
defaultMQProducer.setUseTLS(configure.isUseTLS());
return defaultMQProducer;
}
private TopicList getSystemTopicList() {
RPCHook rpcHook = null;
boolean isEnableAcl = !StringUtils.isEmpty(configure.getAccessKey()) && !StringUtils.isEmpty(configure.getSecretKey());
@@ -305,8 +230,7 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
producer.start();
return producer.getDefaultMQProducerImpl().getmQClientFactory().getMQClientAPIImpl().getSystemTopicList(20000L);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
} finally {
producer.shutdown();
}
@@ -314,61 +238,31 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
@Override
public SendResult sendTopicMessageRequest(SendTopicMessageRequest sendTopicMessageRequest) {
List<TopicConfigInfo> topicConfigInfos = examineTopicConfig(sendTopicMessageRequest.getTopic());
String messageType = topicConfigInfos.get(0).getMessageType();
DefaultMQProducer producer = null;
AclClientRPCHook rpcHook = null;
if (configure.isACLEnabled()) {
rpcHook = new AclClientRPCHook(new SessionCredentials(
configure.getAccessKey(),
configure.getSecretKey()
configure.getAccessKey(),
configure.getSecretKey()
));
}
if (TopicMessageType.TRANSACTION.getValue().equals(messageType)) {
// transaction message
TransactionListener transactionListener = new TransactionListenerImpl();
TransactionMQProducer producer = buildTransactionMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, sendTopicMessageRequest.isTraceEnabled());
producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
producer.setNamesrvAddr(configure.getNamesrvAddr());
producer.setTransactionListener(transactionListener);
try {
producer.start();
Message msg = new Message(sendTopicMessageRequest.getTopic(),
sendTopicMessageRequest.getTag(),
sendTopicMessageRequest.getKey(),
sendTopicMessageRequest.getMessageBody().getBytes()
);
return producer.sendMessageInTransaction(msg, null);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
} finally {
waitSendTraceFinish(producer, sendTopicMessageRequest.isTraceEnabled());
producer.shutdown();
}
} else {
// no transaction message
DefaultMQProducer producer = null;
producer = buildDefaultMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, sendTopicMessageRequest.isTraceEnabled());
producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
producer.setNamesrvAddr(configure.getNamesrvAddr());
try {
producer.start();
Message msg = new Message(sendTopicMessageRequest.getTopic(),
sendTopicMessageRequest.getTag(),
sendTopicMessageRequest.getKey(),
sendTopicMessageRequest.getMessageBody().getBytes()
);
return producer.send(msg);
} catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
} finally {
waitSendTraceFinish(producer, sendTopicMessageRequest.isTraceEnabled());
producer.shutdown();
}
producer = buildDefaultMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, sendTopicMessageRequest.isTraceEnabled());
producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
producer.setNamesrvAddr(configure.getNamesrvAddr());
try {
producer.start();
Message msg = new Message(sendTopicMessageRequest.getTopic(),
sendTopicMessageRequest.getTag(),
sendTopicMessageRequest.getKey(),
sendTopicMessageRequest.getMessageBody().getBytes()
);
return producer.send(msg);
} catch (Exception e) {
throw Throwables.propagate(e);
} finally {
waitSendTraceFinish(producer, sendTopicMessageRequest.isTraceEnabled());
producer.shutdown();
}
}
private void waitSendTraceFinish(DefaultMQProducer producer, boolean traceEnabled) {
@@ -390,20 +284,4 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
} catch (Exception ignore) {
}
}
static class TransactionListenerImpl implements TransactionListener {
private AtomicInteger transactionIndex = new AtomicInteger(0);
private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
return LocalTransactionState.COMMIT_MESSAGE;
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
return LocalTransactionState.COMMIT_MESSAGE;
}
}
}

View File

@@ -24,10 +24,10 @@ import java.util.List;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.dashboard.service.DashboardCollectService;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.apache.rocketmq.tools.admin.MQAdminExt;
@@ -93,8 +93,7 @@ public class CollectTaskRunnble implements Runnable {
try {
list = dashboardCollectService.getTopicMap().get(topic);
} catch (ExecutionException e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
if (null == list) {
list = Lists.newArrayList();

View File

@@ -34,10 +34,10 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import javax.annotation.Resource;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.dashboard.config.RMQConfigure;
import org.apache.rocketmq.dashboard.service.DashboardCollectService;
@@ -84,8 +84,7 @@ public class DashboardCollectTask {
}
}
catch (Exception err) {
Throwables.throwIfUnchecked(err);
throw new RuntimeException(err);
throw Throwables.propagate(err);
}
}
@@ -129,8 +128,7 @@ public class DashboardCollectTask {
log.debug("Broker Collected Data in memory = {}" + JsonUtil.obj2String(dashboardCollectService.getBrokerMap().asMap()));
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -146,12 +144,10 @@ public class DashboardCollectTask {
Thread.sleep(1000);
}
catch (InterruptedException e1) {
Throwables.throwIfUnchecked(e1);
throw new RuntimeException(e1);
throw Throwables.propagate(e1);
}
fetchBrokerRuntimeStats(brokerAddr, retryTime - 1);
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -201,8 +197,7 @@ public class DashboardCollectTask {
}
catch (IOException e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -255,7 +250,7 @@ public class DashboardCollectTask {
private void addSystemTopic() throws Exception {
ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
Map<String, Set<String>> clusterTable = clusterInfo.getClusterAddrTable();
HashMap<String, Set<String>> clusterTable = clusterInfo.getClusterAddrTable();
for (Map.Entry<String, Set<String>> entry : clusterTable.entrySet()) {
String clusterName = entry.getKey();
TopicValidator.addSystemTopic(clusterName);

View File

@@ -42,9 +42,7 @@ rocketmq:
# configure multiple namesrv addresses to manage multiple different clusters
namesrvAddrs:
- 127.0.0.1:9876
# - 127.0.0.2:9876
# - 10.151.47.32:9876;10.151.47.33:9876;10.151.47.34:9876
# - 10.151.47.30:9876
- 127.0.0.2:9876
# if you use rocketmq version < 3.5.8, rocketmq.config.isVIPChannel should be false.default true
isVIPChannel:
# timeout for mqadminExt, default 5000ms
@@ -60,8 +58,8 @@ rocketmq:
loginRequired: false
useTLS: false
# set the accessKey and secretKey if you used acl
# accessKey: rocketmq2
# secretKey: 12345678
accessKey: # if version > 4.4.0
secretKey: # if version > 4.4.0
threadpool:
config:

View File

@@ -45,7 +45,6 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
$scope.writeOperationEnabled = $scope.userRole == null ? true : ($scope.userRole == 1 ? true : false);
$scope.filterNormal = true;
$scope.filterSystem = false;
$scope.filterFIFO = false;
$scope.doSort = function () {// todo how to change this fe's code ? (it's dirty)
if ($scope.sortKey == 'diffTotal') {
@@ -76,10 +75,7 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
$http({
method: "GET",
url: "consumer/groupList.query",
params: {
skipSysGroup: false,
}
url: "consumer/groupList.query"
}).success(function (resp) {
if (resp.status == 0) {
$scope.allConsumerGrouopList = resp.data;
@@ -139,28 +135,16 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
$scope.filterList(1);
});
$scope.$watch('filterFIFO', function () {
$scope.filterList(1);
});
$scope.filterByType = function (str, type,version) {
$scope.filterByType = function (str) {
if ($scope.filterSystem) {
if (type === "SYSTEM") {
if (str.startsWith("%S")) {
return true
}
}
if ($scope.filterNormal) {
if (type === "NORMAL") {
if (str.startsWith("%") == false) {
return true
}
if(!version && type === "FIFO"){
return true;
}
}
if ($scope.filterFIFO) {
if (type === "FIFO") {
return true;
}
}
return false;
};
@@ -170,7 +154,7 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
var canShowList = [];
$scope.allConsumerGrouopList.forEach(function (element) {
console.log(element)
if ($scope.filterByType(element.group, element.subGroupType, $scope.rmqVersion)) {
if ($scope.filterByType(element.group)) {
if (element.group.toLowerCase().indexOf(lowExceptStr) != -1) {
canShowList.push(element);
}
@@ -205,7 +189,6 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
subscriptionGroupConfig: {
groupName: "",
consumeEnable: true,
consumeMessageOrderly: false,
consumeFromMinEnable: true,
consumeBroadcastEnable: true,
retryQueueNums: 1,
@@ -228,7 +211,7 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
// Refresh topic list
$scope.refreshConsumerData();
},
template: $scope.rmqVersion ? 'consumerModifyDialogForV5' : 'consumerModifyDialog',
template: 'consumerModifyDialog',
controller: 'consumerModifyDialogController',
data: {
consumerRequestList: request,

View File

@@ -15,18 +15,10 @@
* limitations under the License.
*/
app.controller('AppCtrl', ['$scope','$window','$translate','$http','Notification', function ($scope,$window,$translate, $http, Notification) {
$scope.rmqVersion = localStorage.getItem("isV5");
$scope.changeTranslate = function(langKey){
$translate.use(langKey);
}
$scope.changeRMQVersion = function (version) {
$scope.rmqVersion = version === 5;
var v = version === 5;
localStorage.setItem("isV5", v);
}
$scope.logout = function(){
$http({
method: "POST",

View File

@@ -50,16 +50,13 @@ var en = {
"ROUTER": "Router",
"MANAGE": "Manage",
"CONFIG": "Config",
"SEND_MSG": "Send Message",
"SEND_MSG": "Send Massage",
"RESET_CUS_OFFSET": "Reset Consumer Offset",
"DELETE": "Delete",
"CHANGE_LANG": "ChangeLanguage",
"CHANGE_VERSION": "ChangeVersion",
"BROKER": "Broker",
"NORMAL": "NORMAL",
"RETRY": "RETRY",
"FIFO": "FIFO",
"TRANSACTION": "TRANSACTION",
"DLQ": "DLQ",
"QUANTITY":"Quantity",
"TYPE":"Type",
@@ -126,11 +123,5 @@ var en = {
"GROUP_PERM":"Group Permission",
"SYNCHRONIZE":"Synchronize Data",
"SHOW":"Show",
"HIDE":"Hide",
"MESSAGE_TYPE":"messageType",
"MESSAGE_TYPE_UNSPECIFIED": "UNSPECIFIED, is NORMAL",
"MESSAGE_TYPE_NORMAL": "NORMAL",
"MESSAGE_TYPE_FIFO": "FIFO",
"MESSAGE_TYPE_DELAY": "DELAY",
"MESSAGE_TYPE_TRANSACTION": "TRANSACTION",
"HIDE":"Hide"
}

View File

@@ -55,12 +55,9 @@ var zh = {
"SKIP_MESSAGE_ACCUMULATE":"跳过堆积",
"DELETE": "删除",
"CHANGE_LANG": "更换语言",
"CHANGE_VERSION": "更换版本",
"BROKER": "Broker",
"NORMAL": "普通",
"RETRY": "重试",
"FIFO": "顺序",
"TRANSACTION": "事务",
"DLQ": "死信",
"QUANTITY":"数量",
"TYPE":"类型",
@@ -127,11 +124,5 @@ var zh = {
"GROUP_PERM":"消费组权限",
"SYNCHRONIZE":"同步",
"SHOW":"显示",
"HIDE":"隐藏",
"MESSAGE_TYPE":"消息类型",
"MESSAGE_TYPE_UNSPECIFIED": "未指定,为普通消息",
"MESSAGE_TYPE_NORMAL": "普通消息",
"MESSAGE_TYPE_FIFO": "顺序消息",
"MESSAGE_TYPE_DELAY": "定时/延时消息",
"MESSAGE_TYPE_TRANSACTION": "事务消息",
"HIDE":"隐藏"
}

View File

@@ -45,30 +45,24 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
}
};
$scope.filterNormal = true
$scope.filterDelay = false
$scope.filterFifo = false
$scope.filterTransaction = false
$scope.filterRetry = false
$scope.filterDLQ = false
$scope.filterSystem = false
$scope.allTopicList = [];
$scope.allTopicNameList = [];
$scope.allMessageTypeList = [];
$scope.topicShowList = [];
$scope.userRole = $window.sessionStorage.getItem("userrole");
$scope.writeOperationEnabled = $scope.userRole == null ? true : ($scope.userRole == 1 ? true : false);
$scope.writeOperationEnabled = $scope.userRole == null ? true : ($scope.userRole == 1 ? true : false);
$scope.refreshTopicList = function () {
$http({
method: "GET",
url: "topic/list.queryTopicType"
url: "topic/list.query"
}).success(function (resp) {
if (resp.status == 0) {
$scope.allTopicNameList = resp.data.topicNameList;
$scope.allMessageTypeList = resp.data.messageTypeList;
console.log($scope.allTopicNameList);
$scope.allTopicList = resp.data.topicList.sort();
console.log($scope.allTopicList);
console.log(JSON.stringify(resp));
$scope.showTopicList(1, $scope.allTopicNameList.length);
$scope.showTopicList(1, $scope.allTopicList.length);
} else {
Notification.error({message: resp.errMsg, delay: 5000});
@@ -85,15 +79,6 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
$scope.$watch('filterNormal', function () {
$scope.filterList(1);
});
$scope.$watch('filterFifo', function () {
$scope.filterList(1);
});
$scope.$watch('filterTransaction', function () {
$scope.filterList(1);
});
$scope.$watch('filterDelay', function () {
$scope.filterList(1);
});
$scope.$watch('filterRetry', function () {
$scope.filterList(1);
});
@@ -107,13 +92,13 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
var lowExceptStr = $scope.filterStr.toLowerCase();
var canShowList = [];
for (let i = 0; i < $scope.allTopicNameList.length; ++i) {
if ($scope.filterByType($scope.allTopicNameList[i], $scope.allMessageTypeList[i])) {
if ($scope.allTopicNameList[i].toLowerCase().indexOf(lowExceptStr) != -1) {
canShowList.push($scope.allTopicNameList[i]);
$scope.allTopicList.forEach(function (element) {
if ($scope.filterByType(element)) {
if (element.toLowerCase().indexOf(lowExceptStr) != -1) {
canShowList.push(element);
}
}
}
});
$scope.paginationConf.totalItems = canShowList.length;
var perPage = $scope.paginationConf.itemsPerPage;
var from = (currentPage - 1) * perPage;
@@ -121,7 +106,7 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
$scope.topicShowList = canShowList.slice(from, to);
};
$scope.filterByType = function (str, type) {
$scope.filterByType = function (str) {
if ($scope.filterRetry) {
if (str.startsWith("%R")) {
return true
@@ -138,22 +123,7 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
}
}
if ($scope.filterNormal) {
if (type.includes("NORMAL")) {
return true
}
}
if ($scope.filterDelay) {
if (type.includes("DELAY")) {
return true
}
}
if ($scope.filterFifo) {
if (type.includes("FIFO")) {
return true
}
}
if ($scope.filterTransaction) {
if (type.includes("TRANSACTION")) {
if (str.startsWith("%") == false) {
return true
}
}
@@ -168,10 +138,10 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
var perPage = $scope.paginationConf.itemsPerPage;
var from = (currentPage - 1) * perPage;
var to = (from + perPage) > totalItem ? totalItem : from + perPage;
console.log($scope.allTopicNameList);
console.log($scope.allTopicList);
console.log(from)
console.log(to)
$scope.topicShowList = $scope.allTopicNameList.slice(from, to);
$scope.topicShowList = $scope.allTopicList.slice(from, to);
$scope.paginationConf.totalItems = totalItem;
console.log($scope.topicShowList)
console.log($scope.paginationConf.totalItems)
@@ -358,8 +328,8 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
var bIsUpdate = true;
if (request == null) {
request = [{
writeQueueNums: 8,
readQueueNums: 8,
writeQueueNums: 16,
readQueueNums: 16,
perm: 6,
order: false,
topicName: "",
@@ -385,7 +355,6 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
topicRequestList: request,
allClusterNameList: Object.keys(resp.data.clusterInfo.clusterAddrTable),
allBrokerNameList: Object.keys(resp.data.brokerServer),
allMessageTypeList: resp.data.messageTypes,
bIsUpdate: bIsUpdate,
writeOperationEnabled: $scope.writeOperationEnabled
}

View File

@@ -47,14 +47,6 @@
<li><a href="javascript:void(0)" ng-click="changeTranslate('zh')">Simplified Chinese</a></li>
</ul>
</li>
<li class="dropdown">
<a href="bootstrap-elements.html" data-target="#" class="dropdown-toggle" data-toggle="dropdown">{{'CHANGE_VERSION' | translate}}
<b class="caret"></b></a>
<ul class="dropdown-menu">
<li><a href="javascript:void(0)" ng-click="changeRMQVersion(5)">RocketMQ 5.x</a></li>
<li><a href="javascript:void(0)" ng-click="changeRMQVersion(4)">RocketMQ 4.x</a></li>
</ul>
</li>
<li class="dropdown" ng-show="username != ''">
<a href="bootstrap-elements.html" data-target="#" class="dropdown-toggle" data-toggle="dropdown">{{username}}
<b class="caret"></b></a>

View File

@@ -50,14 +50,7 @@
<td class="text-center">{{instance.address}}</td>
<td class="text-center">{{instance.brokerVersionDesc}}</td>
<td class="text-center">{{instance.putTps.split(' ')[0]| number:2}}</td>
<td class="text-center">
<span ng-if="!instance.getTransferedTps || !instance.getTransferedTps.trim()">
{{instance.getTransferredTps.split(' ')[0] | number:2}}
</span>
<span ng-if="instance.getTransferedTps && instance.getTransferedTps.trim()">
{{instance.getTransferedTps.split(' ')[0] | number:2}}
</span>
</td>
<td class="text-center">{{instance.getTransferedTps.split(' ')[0]| number:2}}</td>
<td class="text-center">{{instance.msgPutTotalTodayMorning -
instance.msgPutTotalYesterdayMorning}}
</td>

View File

@@ -26,8 +26,6 @@
</div>
<md-checkbox aria-label="Checkbox" ng-model="filterNormal" class="md-primary">{{'NORMAL' | translate}}
</md-checkbox>
<md-checkbox aria-label="Checkbox" ng-show="rmqVersion" ng-model="filterFIFO" class="md-primary">{{'FIFO' | translate}}
</md-checkbox>
<md-checkbox aria-label="Checkbox" ng-model="filterSystem" class="md-primary">{{'SYSTEM' | translate}}
</md-checkbox>
<button class="btn btn-raised btn-sm btn-primary" type="button" ng-show="{{writeOperationEnabled}}"
@@ -322,116 +320,6 @@
</div>
</script>
<script type="text/ng-template" id="consumerModifyDialogForV5">
<div>
<div>
<div class="modal-header">
<h4 class="modal-title">{{'SUBSCRIPTION_CHANGE'|translate}}</h4>
</div>
<div class="modal-body " ng-repeat="item in ngDialogData.consumerRequestList">
<form id="addAppForm1" name="addAppForm" class="form-horizontal" novalidate>
<div class="form-group" ng-hide="ngDialogData.bIsUpdate">
<label class="control-label col-sm-3">clusterName:</label>
<div class="col-sm-9">
<select name="mySelectClusterNameList" multiple chosen
ng-model="item.clusterNameList"
ng-options="clusterNameItem for clusterNameItem in ngDialogData.allClusterNameList">
<option value=""></option>
</select>
</div>
</div>
<div class="form-group">
<label class="control-label col-sm-3">brokerName:</label>
<div class="col-sm-9">
<select name="mySelectBrokerNameList" multiple chosen
ng-disabled="ngDialogData.bIsUpdate"
ng-model="item.brokerNameList"
ng-options="brokerNameItem for brokerNameItem in ngDialogData.allBrokerNameList">
<option value=""></option>
</select>
</div>
</div>
<div class="form-group">
<label class="control-label col-sm-3">groupName:</label>
<div class="col-sm-9">
<input class="form-control" ng-model="item.subscriptionGroupConfig.groupName" type="text"
ng-disabled="ngDialogData.bIsUpdate" required/>
<span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
</div>
</div>
<div class="form-group">
<label class="control-label col-sm-3">consumeEnable:</label>
<div class="col-sm-9">
<md-switch class="md-primary" ng-disabled="{{!ngDialogData.writeOperationEnabled}}" md-no-ink
aria-label="Switch No Ink" ng-model="item.subscriptionGroupConfig.consumeEnable">
</md-switch>
</div>
</div>
<div class="form-group">
<label class="control-label col-sm-3">consumeOrderlyEnable:</label>
<div class="col-sm-9">
<md-switch class="md-primary custom-md-switch" eng-disabled="{{!ngDialogData.writeOperationEnabled}}" md-no-ink
aria-label="Switch No Ink" ng-model="item.subscriptionGroupConfig.consumeMessageOrderly">
</md-switch>
<span style="font-size: 12px;">[Pay Attention: FIFO ConsumerGroup Need Open 'consumeOrderlyEnable' Option]</span>
</div>
</div>
<div class="form-group">
<label class="control-label col-sm-3">consumeBroadcastEnable:</label>
<div class="col-sm-9">
<md-switch class="md-primary" ng-disabled="{{!ngDialogData.writeOperationEnabled}}" md-no-ink
aria-label="Switch No Ink"
ng-model="item.subscriptionGroupConfig.consumeBroadcastEnable">
</md-switch>
</div>
</div>
<div class="form-group">
<label class="control-label col-sm-3">retryQueueNums:</label>
<div class="col-sm-9">
<input class="form-control" ng-model="item.subscriptionGroupConfig.retryQueueNums"
type="text" ng-disabled="{{!ngDialogData.writeOperationEnabled}}"
required/>
<span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
</div>
</div>
<div class="form-group">
<label class="control-label col-sm-3">brokerId:</label>
<div class="col-sm-9">
<input class="form-control" ng-model="item.subscriptionGroupConfig.brokerId" type="text"
ng-disabled="{{!ngDialogData.writeOperationEnabled}}" required/>
<span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
</div>
</div>
<div class="form-group">
<label class="control-label col-sm-3">whichBrokerWhenConsumeSlowly:</label>
<div class="col-sm-9">
<input class="form-control"
ng-model="item.subscriptionGroupConfig.whichBrokerWhenConsumeSlowly" type="text"
ng-disabled="{{!ngDialogData.writeOperationEnabled}}" required/>
<span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
</div>
</div>
</form>
<div class="modal-footer">
<div class="ngdialog-buttons">
<button type="button" class="ngdialog-button ngdialog-button-primary"
ng-disabled="addAppForm.$invalid"
ng-show="{{ngDialogData.writeOperationEnabled}}"
ng-click="postConsumerRequest(item)">{{ 'COMMIT' | translate }}
</button>
<button type="button" class="ngdialog-button ngdialog-button-secondary"
ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
</button>
</div>
</div>
</div>
</div>
</div>
</div>
</script>
<!--consumer monitor config-->
<script type="text/ng-template" id="consumerMonitorDialog">
<div class="modal-header">

View File

@@ -24,12 +24,6 @@
</div>
<md-checkbox aria-label="Checkbox" ng-model="filterNormal" class="md-primary">{{'NORMAL' | translate}}
</md-checkbox>
<md-checkbox aria-label="Checkbox" ng-model="filterDelay" class="md-primary" ng-show="rmqVersion">{{'DELAY' | translate}}
</md-checkbox>
<md-checkbox aria-label="Checkbox" ng-model="filterFifo" class="md-primary" ng-show="rmqVersion">{{'FIFO' | translate}}
</md-checkbox>
<md-checkbox aria-label="Checkbox" ng-model="filterTransaction" class="md-primary" ng-show="rmqVersion">{{'TRANSACTION' | translate}}
</md-checkbox>
<md-checkbox aria-label="Checkbox" ng-model="filterRetry" class="md-primary">{{'RETRY' | translate}}
</md-checkbox>
<md-checkbox aria-label="Checkbox" ng-model="filterDLQ" class="md-primary">{{'DLQ' | translate}}
@@ -69,7 +63,6 @@
<button class="btn btn-raised btn-sm btn-primary" type="button"
ng-click="openUpdateDialog(topic, sysFlag)">topic {{'CONFIG' |translate}}
</button>
<!-- todo 发送消息,根据消息类型判断-->
<button class="btn btn-raised btn-sm btn-primary" type="button"
ng-show="{{!sysFlag}}"
ng-click="openSendTopicMessageDialog(topic)">{{'SEND_MSG' | translate}}
@@ -196,18 +189,6 @@
<span class="text-danger" ng-show="addAppForm.topicName.$error.required">{{'TOPIC_NAME'|translate}}不能为空.</span>
</div>
</div>
<!-- 设置topic 类型 -->
<div class="form-group">
<label class="control-label col-sm-2">{{'MESSAGE_TYPE'|translate}}:</label>
<div class="col-sm-10">
<select name="mySelectMessageType" chosen ng-disabled="ngDialogData.bIsUpdate"
ng-model="item.messageType"
ng-options="messageType as value | translate disable when messageType=='UNSPECIFIED' for (messageType , value) in ngDialogData.allMessageTypeList"
>
<option value=""></option>
</select>
</div>
</div>
<div class="form-group">
<label class="control-label col-sm-2">{{'WRITE_QUEUE_NUMS'|translate}}:</label>
<div class="col-sm-10">

View File

@@ -34,27 +34,27 @@ import org.apache.rocketmq.client.impl.MQClientAPIImpl;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.admin.RollbackStats;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.body.ProducerConnection;
import org.apache.rocketmq.common.protocol.body.QueueTimeSpan;
import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.dashboard.service.client.MQAdminExtImpl;
import org.apache.rocketmq.dashboard.service.client.MQAdminInstance;
import org.apache.rocketmq.dashboard.util.MockObjectUtil;
@@ -82,7 +82,6 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyMap;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
@@ -571,13 +570,11 @@ public class MQAdminExtImplTest {
public void testCreateTopic() throws Exception {
assertNotNull(mqAdminExtImpl);
{
doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt(), anyMap());
doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt(), anyInt(), anyMap());
doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt());
doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt(), anyInt());
}
Map<String, String> map = new HashMap<>();
map.put("message.type", "FIFO");
mqAdminExtImpl.createTopic("key", "topic_test", 8, map);
mqAdminExtImpl.createTopic("key", "topic_test", 8, 1, map);
mqAdminExtImpl.createTopic("key", "topic_test", 8);
mqAdminExtImpl.createTopic("key", "topic_test", 8, 1);
}
@Test

View File

@@ -18,7 +18,7 @@ package org.apache.rocketmq.dashboard.admin;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.dashboard.config.RMQConfigure;
import org.apache.rocketmq.dashboard.util.MockObjectUtil;
import org.apache.rocketmq.tools.admin.MQAdminExt;

View File

@@ -21,7 +21,7 @@ import com.google.common.collect.Lists;
import java.util.List;
import org.apache.rocketmq.common.AclConfig;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.dashboard.model.request.AclRequest;
import org.apache.rocketmq.dashboard.service.impl.AclServiceImpl;
import org.apache.rocketmq.dashboard.util.MockObjectUtil;

View File

@@ -18,8 +18,8 @@ package org.apache.rocketmq.dashboard.controller;
import java.util.HashMap;
import java.util.Properties;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.dashboard.service.impl.ClusterServiceImpl;
import org.apache.rocketmq.dashboard.util.MockObjectUtil;
import org.junit.Test;

View File

@@ -23,17 +23,17 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.admin.RollbackStats;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
import org.apache.rocketmq.dashboard.model.request.DeleteSubGroupRequest;
import org.apache.rocketmq.dashboard.model.request.ResetOffsetRequest;

View File

@@ -21,10 +21,10 @@ import com.google.common.collect.Lists;
import java.util.List;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.CMResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.body.CMResult;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.dashboard.model.DlqMessageRequest;
import org.apache.rocketmq.dashboard.model.MessagePage;
import org.apache.rocketmq.dashboard.model.MessageView;

View File

@@ -32,9 +32,9 @@ import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.body.CMResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.CMResult;
import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.dashboard.model.QueueOffsetInfo;
import org.apache.rocketmq.dashboard.model.request.MessageQuery;
import org.apache.rocketmq.dashboard.service.impl.MessageServiceImpl;

View File

@@ -19,8 +19,8 @@ package org.apache.rocketmq.dashboard.controller;
import java.util.HashSet;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
import org.apache.rocketmq.common.protocol.body.Connection;
import org.apache.rocketmq.common.protocol.body.ProducerConnection;
import org.apache.rocketmq.dashboard.interceptor.AuthInterceptor;
import org.apache.rocketmq.dashboard.service.impl.LoginServiceImpl;
import org.apache.rocketmq.dashboard.service.impl.ProducerServiceImpl;

View File

@@ -29,16 +29,16 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
import org.apache.rocketmq.dashboard.service.impl.ConsumerServiceImpl;

View File

@@ -33,12 +33,12 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.GroupList;
import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.KVTable;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.dashboard.BaseTest;
import org.apache.rocketmq.dashboard.config.CollectExecutorConfig;
import org.apache.rocketmq.dashboard.config.RMQConfigure;

View File

@@ -29,7 +29,7 @@ import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
import org.apache.rocketmq.dashboard.service.ConsumerService;
@@ -78,8 +78,7 @@ public abstract class RocketMQConsoleTestBase {
}
}
}
Throwables.throwIfUnchecked(exception);
throw new RuntimeException(exception);
throw Throwables.propagate(exception);
}
}
@@ -92,8 +91,7 @@ public abstract class RocketMQConsoleTestBase {
producer.start();
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}
@@ -139,8 +137,7 @@ public abstract class RocketMQConsoleTestBase {
consumer.start();
}
catch (Exception e) {
Throwables.throwIfUnchecked(e);
throw new RuntimeException(e);
throw Throwables.propagate(e);
}
}

View File

@@ -33,39 +33,39 @@ import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.trace.TraceConstants;
import org.apache.rocketmq.client.trace.TraceType;
import org.apache.rocketmq.common.AclConfig;
import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.apache.rocketmq.common.DataVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.admin.OffsetWrapper;
import org.apache.rocketmq.common.admin.TopicOffset;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
import org.apache.rocketmq.remoting.protocol.body.BrokerStatsItem;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ConsumeStatus;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.remoting.protocol.body.ProcessQueueInfo;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.QueueData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
import org.apache.rocketmq.common.protocol.body.BrokerStatsItem;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.Connection;
import org.apache.rocketmq.common.protocol.body.ConsumeStatus;
import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.common.protocol.body.ProcessQueueInfo;
import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.QueueData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.dashboard.model.DlqMessageRequest;
import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.checkerframework.checker.units.qual.A;
import static org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType.CONSUME_ACTIVELY;
import static org.apache.rocketmq.common.protocol.heartbeat.ConsumeType.CONSUME_ACTIVELY;
public class MockObjectUtil {