[Bug][Manager] Problems arising from the MQ cluster during deletion process
- Dominant language
- Java
- Stars
- 1.5k
- Forks
- 572
- Avg merge
- 22h 7m
- Merged PRs (30d)
- 8
Description
### What happened
When creating an access in the data access module, you can select the corresponding MQ type to create a data flow group.
Even without a corresponding MQ cluster, new connections can still be established and submitted for approval. However, issues may arise when deleting because there is no corresponding MQ cluster.
Furthermore, I believe that if there is no corresponding MQ cluster, the creation should not be successful.
### What you expected to happen
Deletion process
InlongGroupController.delete(...)
-> InlongGroupProcessService.deleteProcess(String groupId, String operator)
-> InlongGroupProcessService.invokeDeleteProcess(...)
-> workflowService.start(ProcessName.DELETE_GROUP_PROCESS, operator, form)
The workflow will call DeleteGroupWorkflowDefinition.defineProcess(), which includes the DeleteMQ task in the deletion process.
This task will execute QueueResourceListener.listen(...)
In the DELETE case, it calls queueOperator.deleteQueueForGroup(groupInfo, operator);
This method has different implementations for different MQs.
In Kafka
```java
ClusterInfo clusterInfo = clusterService.getOne(groupInfo.getInlongClusterTag(), null, ClusterType.KAFKA);
@Override
public ClusterInfo getOne(String clusterTag, String name, String type) {
List entityList = clusterMapper.selectByKey(clusterTag, name, type);
if (CollectionUtils.isEmpty(entityList)) {
throw new BusinessException(String.format("cluster not found by tag=%s, name=%s, type=%s",
clusterTag, name, type));
}
InlongClusterEntity entity = entityList.get(0);
InlongClusterOperator instance = clusterOperatorFactory.getInstance(entity.getType());
ClusterInfo result = instance.getFromEntity(entity);
LOGGER.debug("success to get inlong cluster by tag={}, name={}, type={}", clusterTag, name, type);
return result;
}
```
An exception will be thrown if no cluster of the corresponding type is found.
In the log
```
[ ] 2026-03-30 05:53:58.572 -ERROR [http-nio-8083-exec-10] a.i.m.w.e.LogableEventListener:88 - execute listener WorkflowEventLogEntity(id=null, processId=48, processName=DELETE_GROUP_PROCESS, processDisplayName=Delete Group, inlongGroupId=test_group_id, taskId=130, elementName=DeleteMQ, elementDisplayName=Group-DeleteMQ, eventType=TaskEvent, event=COMPLETE, listener=QueueResourceListener, startTime=Mon Mar 30 05:53:58 UTC 2026, endTime=null, status=-1, async=0, ip=172.18.0.7, remark=null, exception=cluster not found by tag=null, name=null, type=KAFKA) error:
org.apache.inlong.manager.common.exceptions.BusinessException: cluster not found by tag=null, name=null, type=KAFKA
at org.apache.inlong.manager.service.cluster.InlongClusterServiceImpl.getOne(InlongClusterServiceImpl.java:545) ~[manager-service-2.3.0.jar:2.3.0]
```
In Pulsar
```java
List clusterInfos = clusterService.listByTagAndType(clusterTag, ClusterType.PULSAR);
@Override
public List listByTagAndType(String clusterTag, String clusterType) {
List clusterEntities = clusterMapper.selectByKey(clusterTag, null, clusterType);
if (CollectionUtils.isEmpty(clusterEntities)) {
throw new BusinessException(String.format("cannot find any cluster by tag %s and type %s",
clusterTag, clusterType));
}
List clusterInfos = clusterEntities.stream()
.map(entity -> {
InlongClusterOperator operator = clusterOperatorFactory.getInstance(entity.getType());
return operator.getFromEntity(entity);
})
.collect(Collectors.toList());
LOGGER.debug("success to list inlong cluster by tag={}", clusterTag);
return clusterInfos;
}
```
The TubeMQ deletion process has not been implemented yet, so no exception will be triggered.
### How to reproduce
1. Create a new connection and use an MQ without a corresponding MQ cluster.
2. Delete it
### Environment
_No response_
### InLong version
2.3.0
### InLong Component
InLong Manager
### Are you willing to submit PR?
- [x] Yes, I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
Contributor guide
Research direction
Start at InlongGroupController.delete(...) and follow InlongGroupProcessService.deleteProcess(...) into DeleteGroupWorkflowDefinition.defineProcess(). Read QueueResourceListener.listen(...) and the Kafka and Pulsar queueOperator.deleteQueueForGroup(...) paths, including the clusterService calls. Done means creation is rejected when its MQ cluster is missing and deletion no longer fails because that cluster cannot be found.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100