Skip to content

[INLONG-12106][Manager] Add relevant MQ cluster checks #12107

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
flowers-59f wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[INLONG-12106][Manager] Add relevant MQ cluster checks #12107

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 21 additions & 1 deletion ...service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -215,7 +215,27 @@ public String save(InlongGroupRequest request, String operator) {
if (request.getEnableZookeeper() == null) {
request.setEnableZookeeper(enableZookeeper ? InlongConstants.ENABLE_ZK : InlongConstants.DISABLE_ZK);
}
InlongGroupOperator instance = groupOperatorFactory.getInstance(request.getMqType());

if (request.getEnableCreateResource() == null) {
request.setEnableCreateResource(InlongConstants.ENABLE_CREATE_RESOURCE);
}

if (request.getInlongGroupMode() == null) {
request.setInlongGroupMode(InlongConstants.STANDARD_MODE);
Copy link
Copy Markdown
Contributor

@fuweng11 fuweng11 May 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is no need to set a default value here, as the create table statement for the inlong_group table has already set a default value

Copy link
Copy Markdown
Contributor Author

@flowers-59f flowers-59f May 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In QueueResourceListener.listen()

        if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())
                || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())) {
            log.warn("skip to execute QueueResourceListener as sync mode {} (1 for realtime sync, 2 for offline sync) "
                    + "for groupId={}", groupInfo.getInlongGroupMode(), groupId);
            if (GroupOperateType.INIT.equals(operateType)) {
                this.createQueueForStreams(groupInfo, groupProcessForm.getStreamInfos(), operator);
            }
            return ListenerResult.success("skip - disable create mq resource for sync mode");
        }

        if (InlongConstants.DISABLE_CREATE_RESOURCE.equals(groupInfo.getEnableCreateResource())) {
            log.warn("skip to execute QueueResourceListener as disable create resource for groupId={}", groupId);
            return ListenerResult.success("skip - disable create resource");
        }

In other modes, the creation/deletion of MQ resources will be skipped, so we only need to check for the existence of the corresponding MQ in specific modes. But, under this save link, the EnableCreateResource and InlongGroupMode passed in the request do not have values, so I set them according to the default values and made a check. Since there are other places that call this function, they might have set the corresponding values, and in those cases, validation might not be required.

This is my idea. Please take a look and see if there are any issues and let me know how I can improve.Thank you very much.

Copy link
Copy Markdown
Contributor

@fuweng11 fuweng11 May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When saving the group, no clustertag is assigned, so this verification is unnecessary. Additionally, since the inlong_group table has default values set, there is no need to explicitly set them in the code during saving.
企业微信截图_ab049c8e-93eb-4d43-bcea-80d464e84832

}

String mqType = request.getMqType();
if (InlongConstants.STANDARD_MODE.equals(request.getInlongGroupMode())
Copy link
Copy Markdown
Contributor

@fuweng11 fuweng11 May 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When saving the group, no inlong_cluster_tag was assigned, so there is no need for judgment here.

&& InlongConstants.ENABLE_CREATE_RESOURCE.equals(request.getEnableCreateResource())) {
String clusterTag = request.getInlongClusterTag();
List<InlongClusterEntity> clusterEntities = clusterEntityMapper.selectByKey(clusterTag, null, mqType);
if (CollectionUtils.isEmpty(clusterEntities)) {
throw new BusinessException(String.format("cannot find any cluster by tag %s and type %s",
clusterTag, mqType));
}
}

InlongGroupOperator instance = groupOperatorFactory.getInstance(mqType);
groupId = instance.saveOpt(request, operator);

// save ext info
Expand Down
Loading