You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by "healchow (via GitHub)" <gi...@apache.org> on 2026/02/16 05:49:31 UTC
[PR] [INLONG-12073][Manager] Optimize Pulsar message query with thread pool and proper error handling [inlong]
healchow opened a new pull request, #12074:
URL: https://github.com/apache/inlong/pull/12074
[INLONG-12073][Manager] Optimize Pulsar message query with thread pool and proper error handling
Fixes #12073
### Motivation
Currently, when querying MQ messages from multiple Pulsar clusters, there are several issues:
1. **No thread pool management**: Message queries are executed without proper thread pool management, which may lead to resource exhaustion under high concurrency.
2. **No graceful handling for task rejection**: When too many concurrent requests come in, the system doesn't properly handle the `RejectedExecutionException` and doesn't provide a user-friendly error response.
3. **No task cancellation mechanism**: When task submission fails, previously submitted tasks continue to run unnecessarily.
4. **No interruption support**: Long-running IO operations cannot be cancelled when the request is aborted.
### Modifications
1. **Add a dedicated thread pool**: Use `ThreadPoolTaskExecutor` with configurable core/max pool size and queue capacity for message query tasks.
2. **Add proper error handling**:
- Add a new error code `PULSAR_QUERY_REJECTED(2610)` for task rejection scenarios
- Return proper error code and message to the caller: `{"success": false, "errCode": 2610, "errMsg": "Query task rejected: too many concurrent requests, please try again later"}`
3. **Implement task cancellation**: When `RejectedExecutionException` occurs, cancel all previously submitted tasks to free up resources.
4. **Add interruption checks**: Check `Thread.currentThread().isInterrupted()` before and after IO operations to support task cancellation.
### Verifying this change
- [ ] This change is a trivial rework/code cleanup without any test coverage.
- [ ] This change is already covered by existing tests, such as:
*(please describe tests)*
- [x] This change added tests and can be verified as follows:
- *Added unit tests QueryLatestMessagesRunnableTest to verified this change*
### Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
- If a feature is not applicable for documentation, explain why?
- If a feature is not documented yet in this PR, please create a follow-up issue for adding the documentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-12073][Manager] Optimize Pulsar message query with thread pool and proper error handling [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #12074:
URL: https://github.com/apache/inlong/pull/12074#discussion_r2844113665
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java:
##########
@@ -307,34 +372,150 @@ private void deletePulsarTopic(InlongPulsarInfo pulsarInfo, PulsarClusterInfo pu
* Query latest message from pulsar
*/
public List<BriefMQMessage> queryLatestMessages(InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
- QueryMessageRequest request) throws Exception {
- List<ClusterInfo> pulsarClusterList = clusterService.listByTagAndType(groupInfo.getInlongClusterTag(),
- ClusterType.PULSAR);
- List<BriefMQMessage> briefMQMessages = Collections.synchronizedList(new ArrayList<>());
- QueryCountDownLatch queryLatch = new QueryCountDownLatch(request.getMessageCount(), pulsarClusterList.size());
+ QueryMessageRequest request) {
+ String groupId = streamInfo.getInlongGroupId();
+ String clusterTag = groupInfo.getInlongClusterTag();
+ List<ClusterInfo> clusterInfos = clusterService.listByTagAndType(clusterTag, ClusterType.PULSAR);
+ if (CollectionUtils.isEmpty(clusterInfos)) {
+ log.warn("No pulsar cluster found for clusterTag={} for groupId={}", clusterTag, groupId);
+ return Collections.emptyList();
+ }
+
+ // Select clusters and calculate per-cluster query count
+ Integer requestCount = request.getMessageCount();
+ List<ClusterInfo> selectedClusters = randomSelectQueryClusters(clusterInfos, requestCount);
Review Comment:
Data may be concentrated on a specific pulsar, making it impossible for other pulsar clusters to retrieve the data. Therefore, it is not feasible to randomly select a cluster for querying. Instead, all pulsar clusters need to be queried.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-12073][Manager] Optimize Pulsar message query with thread pool and proper error handling [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #12074:
URL: https://github.com/apache/inlong/pull/12074#discussion_r2845216832
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java:
##########
@@ -307,34 +370,127 @@ private void deletePulsarTopic(InlongPulsarInfo pulsarInfo, PulsarClusterInfo pu
* Query latest message from pulsar
*/
public List<BriefMQMessage> queryLatestMessages(InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
- QueryMessageRequest request) throws Exception {
- List<ClusterInfo> pulsarClusterList = clusterService.listByTagAndType(groupInfo.getInlongClusterTag(),
- ClusterType.PULSAR);
- List<BriefMQMessage> briefMQMessages = Collections.synchronizedList(new ArrayList<>());
- QueryCountDownLatch queryLatch = new QueryCountDownLatch(request.getMessageCount(), pulsarClusterList.size());
+ QueryMessageRequest request) {
+ String groupId = streamInfo.getInlongGroupId();
+ String clusterTag = groupInfo.getInlongClusterTag();
+ List<ClusterInfo> clusterInfos = clusterService.listByTagAndType(clusterTag, ClusterType.PULSAR);
+ if (CollectionUtils.isEmpty(clusterInfos)) {
+ log.warn("No pulsar cluster found for clusterTag={} for groupId={}", clusterTag, groupId);
+ return Collections.emptyList();
+ }
+
+ // Select clusters and calculate per-cluster query count
+ Integer requestCount = request.getMessageCount();
+ int clusterSize = clusterInfos.size();
+ int perClusterCount = calculatePerClusterCount(requestCount, clusterSize);
Review Comment:
Since data may be concentrated on a certain cluster, the average should not be calculated based on the number of clusters and queries during querying.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-12073][Manager] Optimize Pulsar message query with thread pool and proper error handling [inlong]
Posted by "healchow (via GitHub)" <gi...@apache.org>.
healchow merged PR #12074:
URL: https://github.com/apache/inlong/pull/12074
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-12073][Manager] Optimize Pulsar message query with thread pool and proper error handling [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #12074:
URL: https://github.com/apache/inlong/pull/12074#discussion_r2844113665
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java:
##########
@@ -307,34 +372,150 @@ private void deletePulsarTopic(InlongPulsarInfo pulsarInfo, PulsarClusterInfo pu
* Query latest message from pulsar
*/
public List<BriefMQMessage> queryLatestMessages(InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
- QueryMessageRequest request) throws Exception {
- List<ClusterInfo> pulsarClusterList = clusterService.listByTagAndType(groupInfo.getInlongClusterTag(),
- ClusterType.PULSAR);
- List<BriefMQMessage> briefMQMessages = Collections.synchronizedList(new ArrayList<>());
- QueryCountDownLatch queryLatch = new QueryCountDownLatch(request.getMessageCount(), pulsarClusterList.size());
+ QueryMessageRequest request) {
+ String groupId = streamInfo.getInlongGroupId();
+ String clusterTag = groupInfo.getInlongClusterTag();
+ List<ClusterInfo> clusterInfos = clusterService.listByTagAndType(clusterTag, ClusterType.PULSAR);
+ if (CollectionUtils.isEmpty(clusterInfos)) {
+ log.warn("No pulsar cluster found for clusterTag={} for groupId={}", clusterTag, groupId);
+ return Collections.emptyList();
+ }
+
+ // Select clusters and calculate per-cluster query count
+ Integer requestCount = request.getMessageCount();
+ List<ClusterInfo> selectedClusters = randomSelectQueryClusters(clusterInfos, requestCount);
Review Comment:
Data may be concentrated on a specific pulsar, making it impossible for other pulsar clusters to retrieve the data. Therefore, it is not feasible to randomly select a cluster for querying. Instead, all pulsar clusters need to be queried.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-12073][Manager] Optimize Pulsar message query with thread pool and proper error handling [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #12074:
URL: https://github.com/apache/inlong/pull/12074#discussion_r2844114747
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java:
##########
@@ -307,34 +372,150 @@ private void deletePulsarTopic(InlongPulsarInfo pulsarInfo, PulsarClusterInfo pu
* Query latest message from pulsar
*/
public List<BriefMQMessage> queryLatestMessages(InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
- QueryMessageRequest request) throws Exception {
- List<ClusterInfo> pulsarClusterList = clusterService.listByTagAndType(groupInfo.getInlongClusterTag(),
- ClusterType.PULSAR);
- List<BriefMQMessage> briefMQMessages = Collections.synchronizedList(new ArrayList<>());
- QueryCountDownLatch queryLatch = new QueryCountDownLatch(request.getMessageCount(), pulsarClusterList.size());
+ QueryMessageRequest request) {
+ String groupId = streamInfo.getInlongGroupId();
+ String clusterTag = groupInfo.getInlongClusterTag();
+ List<ClusterInfo> clusterInfos = clusterService.listByTagAndType(clusterTag, ClusterType.PULSAR);
+ if (CollectionUtils.isEmpty(clusterInfos)) {
+ log.warn("No pulsar cluster found for clusterTag={} for groupId={}", clusterTag, groupId);
+ return Collections.emptyList();
+ }
+
+ // Select clusters and calculate per-cluster query count
+ Integer requestCount = request.getMessageCount();
+ List<ClusterInfo> selectedClusters = randomSelectQueryClusters(clusterInfos, requestCount);
Review Comment:
Data may be concentrated on a specific pulsar, making it impossible for other pulsar clusters to retrieve the data. Therefore, it is not feasible to randomly select a cluster for querying. Instead, all pulsar clusters need to be queried.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-12073][Manager] Optimize Pulsar message query with thread pool and proper error handling [inlong]
Posted by "healchow (via GitHub)" <gi...@apache.org>.
healchow commented on code in PR #12074:
URL: https://github.com/apache/inlong/pull/12074#discussion_r2844307921
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java:
##########
@@ -307,34 +372,150 @@ private void deletePulsarTopic(InlongPulsarInfo pulsarInfo, PulsarClusterInfo pu
* Query latest message from pulsar
*/
public List<BriefMQMessage> queryLatestMessages(InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
- QueryMessageRequest request) throws Exception {
- List<ClusterInfo> pulsarClusterList = clusterService.listByTagAndType(groupInfo.getInlongClusterTag(),
- ClusterType.PULSAR);
- List<BriefMQMessage> briefMQMessages = Collections.synchronizedList(new ArrayList<>());
- QueryCountDownLatch queryLatch = new QueryCountDownLatch(request.getMessageCount(), pulsarClusterList.size());
+ QueryMessageRequest request) {
+ String groupId = streamInfo.getInlongGroupId();
+ String clusterTag = groupInfo.getInlongClusterTag();
+ List<ClusterInfo> clusterInfos = clusterService.listByTagAndType(clusterTag, ClusterType.PULSAR);
+ if (CollectionUtils.isEmpty(clusterInfos)) {
+ log.warn("No pulsar cluster found for clusterTag={} for groupId={}", clusterTag, groupId);
+ return Collections.emptyList();
+ }
+
+ // Select clusters and calculate per-cluster query count
+ Integer requestCount = request.getMessageCount();
+ List<ClusterInfo> selectedClusters = randomSelectQueryClusters(clusterInfos, requestCount);
Review Comment:
Ok, I'll remove the random method later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org