[INLONG-10204][Manager] Kafka sink supports automatic allocation of s…

…ort standalone cluster (#10207)

Co-authored-by: castorqin <qhj00725@qq.com>

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/kafka/KafkaResourceOperator.java

@@ -24,7 +24,7 @@
 import org.apache.inlong.manager.common.util.Preconditions;
 import org.apache.inlong.manager.pojo.sink.SinkInfo;
 import org.apache.inlong.manager.pojo.sink.kafka.KafkaSinkDTO;
-import org.apache.inlong.manager.service.resource.sink.SinkResourceOperator;
+import org.apache.inlong.manager.service.resource.sink.AbstractStandaloneSinkResourceOperator;
 import org.apache.inlong.manager.service.sink.StreamSinkService;
 
 import org.apache.kafka.clients.admin.Admin;
@@ -47,7 +47,7 @@
  * Kafka resource operator for creating Kafka topic
  */
 @Service
-public class KafkaResourceOperator implements SinkResourceOperator {
+public class KafkaResourceOperator extends AbstractStandaloneSinkResourceOperator {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaResourceOperator.class);
 
@@ -76,7 +76,6 @@ public void createSinkResource(SinkInfo sinkInfo) {
                         new NewTopic(topicName, Optional.of(partitionNum), Optional.empty())));
                 result.values().get(topicName).get();
             }
-
             sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_SUCCESSFUL.getCode(),
                     "create kafka topic success");
             LOGGER.info("success to create kafka topic [{}] for sinkInfo={}", topicName, sinkInfo);
@@ -85,6 +84,7 @@ public void createSinkResource(SinkInfo sinkInfo) {
             sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_FAILED.getCode(), e.getMessage());
             throw new WorkflowException("create kafka topic failed, reason: " + e.getMessage());
         }
+        this.assignCluster(sinkInfo);
     }
 
     /**