apache / apache/inlong

[Bug][Manager] Problems arising from the MQ cluster during deletion process

Open
#12,106 1 comment 0 reactions 0 assignees View on GitHub
stage/stale type/bug
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.
Image
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.
Image
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.