Flux<DataEntityList> export(KafkaCluster cluster) {
    return kafkaConnectService.getConnects(cluster)
        .flatMap(connect -> kafkaConnectService.getConnectorNamesWithErrorsSuppress(cluster, connect.getName())
            .flatMap(connectorName -> kafkaConnectService.getConnector(cluster, connect.getName(), connectorName))
            .flatMap(connectorDTO ->
                kafkaConnectService.getConnectorTopics(cluster, connect.getName(), connectorDTO.getName())
                    .map(topics -> createConnectorDataEntity(cluster, connect, connectorDTO, topics)))
            .buffer(100)
            .map(connectDataEntities -> {
              String dsOddrn = Oddrn.connectDataSourceOddrn(connect.getAddress());
              return new DataEntityList()
                  .dataSourceOddrn(dsOddrn)
                  .items(connectDataEntities);
            })
        );
  }
@Test
  void exportsConnectorsAsDataTransformers() {
    ConnectDTO connect = new ConnectDTO();
    connect.setName("testConnect");
    connect.setAddress("http://kconnect:8083");

    ConnectorDTO sinkConnector = new ConnectorDTO();
    sinkConnector.setName("testSink");
    sinkConnector.setType(ConnectorTypeDTO.SINK);
    sinkConnector.setConnect(connect.getName());
    sinkConnector.setConfig(
        Map.of(
            "connector.class", "FileStreamSink",
            "file", "filePathHere",
            "topic", "inputTopic"
        )
    );

    ConnectorDTO sourceConnector = new ConnectorDTO();
    sourceConnector.setName("testSource");
    sourceConnector.setConnect(connect.getName());
    sourceConnector.setType(ConnectorTypeDTO.SOURCE);
    sourceConnector.setConfig(
        Map.of(
            "connector.class", "FileStreamSource",
            "file", "filePathHere",
            "topic", "outputTopic"
        )
    );

    when(kafkaConnectService.getConnects(CLUSTER))
        .thenReturn(Flux.just(connect));

    when(kafkaConnectService.getConnectorNamesWithErrorsSuppress(CLUSTER, connect.getName()))
        .thenReturn(Flux.just(sinkConnector.getName(), sourceConnector.getName()));

    when(kafkaConnectService.getConnector(CLUSTER, connect.getName(), sinkConnector.getName()))
        .thenReturn(Mono.just(sinkConnector));

    when(kafkaConnectService.getConnector(CLUSTER, connect.getName(), sourceConnector.getName()))
        .thenReturn(Mono.just(sourceConnector));

    when(kafkaConnectService.getConnectorTopics(CLUSTER, connect.getName(), sourceConnector.getName()))
        .thenReturn(Mono.just(new ConnectorTopics().topics(List.of("outputTopic"))));

    when(kafkaConnectService.getConnectorTopics(CLUSTER, connect.getName(), sinkConnector.getName()))
        .thenReturn(Mono.just(new ConnectorTopics().topics(List.of("inputTopic"))));

    StepVerifier.create(exporter.export(CLUSTER))
        .assertNext(dataEntityList -> {
          assertThat(dataEntityList.getDataSourceOddrn())
              .isEqualTo("//kafkaconnect/host/kconnect:8083");

          assertThat(dataEntityList.getItems())
              .hasSize(2);

          assertThat(dataEntityList.getItems())
              .filteredOn(DataEntity::getOddrn, "//kafkaconnect/host/kconnect:8083/connectors/testSink")
              .singleElement()
              .satisfies(sink -> {
                assertThat(sink.getMetadata().get(0).getMetadata())
                    .containsOnlyKeys("type", "connector.class", "file", "topic");
                assertThat(sink.getDataTransformer().getInputs()).contains(
                    "//kafka/cluster/localhost:9092/topics/inputTopic");
              });

          assertThat(dataEntityList.getItems())
              .filteredOn(DataEntity::getOddrn, "//kafkaconnect/host/kconnect:8083/connectors/testSource")
              .singleElement()
              .satisfies(source -> {
                assertThat(source.getMetadata().get(0).getMetadata())
                    .containsOnlyKeys("type", "connector.class", "file", "topic");
                assertThat(source.getDataTransformer().getOutputs()).contains(
                    "//kafka/cluster/localhost:9092/topics/outputTopic");
              });

        })
        .verifyComplete();
  }