[Bug][DataProxy] The topic name generated by dataproxy is incorrect

[fuweng11]

What happened

The topic name generated by dataproxy is incorrect.
The following is the logic for creating kakfa topic on the manager side:
1.Get streamInfo.mqResource.
2.If streamInfo.mqResource equal to streamId, set topic to groupInfo.mqResource.streamInfo.mqResource

What you expected to happen

The topic name generated by dataproxy is incorrect.
The following is the logic for creating kakfa topic on the manager side:
1.Get streamInfo.mqResource.
2.If streamInfo.mqResource equal to streamId, set topic to groupInfo.mqResource.streamInfo.mqResource

How to reproduce

Set streamInfo.mqResource to rain streamId inconsistent.

Environment

No response

InLong version

master

InLong Component

InLong DataProxy

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[gosonzhang]

@fuweng11, I don't understand, the implementation[1]

is actually directly migrated [2]:

Is there a problem with the previous logic, then why is there no error reported?

1. https://github.com/apache/inlong/blob/master/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/MetaConfigHolder.java#L375
2. https://github.com/apache/inlong/blob/branch-1.7/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/kafka/KafkaHandler.java#L150

[fuweng11]

@fuweng11, I don't understand, the implementation[1]

is actually directly migrated [2]:

Is there a problem with the previous logic, then why is there no error reported?

1. https://github.com/apache/inlong/blob/master/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/MetaConfigHolder.java#L375
2. https://github.com/apache/inlong/blob/branch-1.7/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/kafka/KafkaHandler.java#L150

Yes, this problem was not discovered before because if streamInfo.mqResource is not set when the stream is created, the manager side will automatically set streamId to streamInfo.mqResource, and in most cases streamId and streamInfo.mqResource` are the same.

https://github.com/apache/inlong/blob/master/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaResourceOperators.java#L115

[gosonzhang]

@fuweng11, ok, got it