BE: issue 1090 Added topics info to connector endpoint (#1442)

Author: germanosin

api/src/main/java/io/kafbat/ui/mapper/KafkaConnectMapper.java

@@ -2,8 +2,10 @@
 
 import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.connect.model.ClusterInfo;
+import io.kafbat.ui.connect.model.Connector;
 import io.kafbat.ui.connect.model.ConnectorStatusConnector;
 import io.kafbat.ui.connect.model.ConnectorTask;
+import io.kafbat.ui.connect.model.ConnectorTopics;
 import io.kafbat.ui.connect.model.ExpandedConnector;
 import io.kafbat.ui.connect.model.NewConnector;
 import io.kafbat.ui.model.ConnectDTO;
@@ -42,6 +44,14 @@ default ClusterInfo toClient(KafkaConnectState state) {
  @Mapping(target = "connect", ignore = true)
  ConnectorDTO fromClient(io.kafbat.ui.connect.model.Connector connector);
 
+ default ConnectorDTO fromClient(Connector connector, ConnectorTopics topics) {
+ ConnectorDTO connectorDto = this.fromClient(connector);
+ if (topics != null) {
+ return connectorDto.topics(topics.getTopics());
+ }
+ return connectorDto;
+ }
+
  ConnectorStatusDTO fromClient(ConnectorStatusConnector connectorStatus);
 
  @Mapping(target = "status", ignore = true)
@@ -177,6 +187,4 @@ default KafkaConnectState.ConnectorState toScrapeState(InternalConnectorInfo con
  connector.getTopics()
  );
  }
-
-
 }

api/src/main/java/io/kafbat/ui/service/KafkaConnectService.java

@@ -226,8 +226,9 @@ private Mono<Boolean> connectorExists(KafkaCluster cluster, String connectName,
  public Mono<ConnectorDTO> getConnector(KafkaCluster cluster, String connectName,
  String connectorName) {
  return api(cluster, connectName)
- .mono(client -> client.getConnector(connectorName)
- .map(kafkaConnectMapper::fromClient)
+ .mono(client ->
+ Mono.zip(client.getConnector(connectorName), getConnectorTopics(cluster, connectName, connectorName))
+ .map(t -> kafkaConnectMapper.fromClient(t.getT1(), t.getT2()))
  .flatMap(connector ->
  client.getConnectorStatus(connector.getName())
  // status request can return 404 if tasks not assigned yet

contract-typespec/api/kafka-connect.tsp

@@ -225,6 +225,7 @@ model Connector {
  type: ConnectorType;
  status: ConnectorStatus;
  connect: string;
+ topics?: string[];
 }
 
 enum ConnectorAction {

contract/src/main/resources/swagger/kafbat-ui-api.yaml

@@ -3674,6 +3674,10 @@ components:
  $ref: '#/components/schemas/ConnectorStatus'
  connect:
  type: string
+ topics:
+ type: array
+ items:
+ type: string
  required:
  - type
  - status