|
@@ -61,37 +61,22 @@ public class KafkaConnectService {
|
|
|
public Flux<FullConnectorInfoDTO> getAllConnectors(final KafkaCluster cluster,
|
|
|
@Nullable final String search) {
|
|
|
return getConnects(cluster)
|
|
|
- .flatMap(connect -> getConnectorNames(cluster, connect.getName()).map(cn -> Tuples.of(connect.getName(), cn)))
|
|
|
- .flatMap(pair -> getConnector(cluster, pair.getT1(), pair.getT2()))
|
|
|
+ .flatMap(connect ->
|
|
|
+ getConnectorNamesWithErrorsSuppress(cluster, connect.getName())
|
|
|
+ .flatMap(connectorName -> getConnector(cluster, connect.getName(), connectorName)))
|
|
|
.flatMap(connector ->
|
|
|
getConnectorConfig(cluster, connector.getConnect(), connector.getName())
|
|
|
- .map(config -> InternalConnectInfo.builder()
|
|
|
- .connector(connector)
|
|
|
- .config(config)
|
|
|
- .build()
|
|
|
- )
|
|
|
- )
|
|
|
+ .map(config -> InternalConnectInfo.builder().connector(connector).config(config).build()))
|
|
|
.flatMap(connectInfo -> {
|
|
|
ConnectorDTO connector = connectInfo.getConnector();
|
|
|
return getConnectorTasks(cluster, connector.getConnect(), connector.getName())
|
|
|
.collectList()
|
|
|
- .map(tasks -> InternalConnectInfo.builder()
|
|
|
- .connector(connector)
|
|
|
- .config(connectInfo.getConfig())
|
|
|
- .tasks(tasks)
|
|
|
- .build()
|
|
|
- );
|
|
|
+ .map(tasks -> connectInfo.toBuilder().tasks(tasks).build());
|
|
|
})
|
|
|
.flatMap(connectInfo -> {
|
|
|
ConnectorDTO connector = connectInfo.getConnector();
|
|
|
return getConnectorTopics(cluster, connector.getConnect(), connector.getName())
|
|
|
- .map(ct -> InternalConnectInfo.builder()
|
|
|
- .connector(connector)
|
|
|
- .config(connectInfo.getConfig())
|
|
|
- .tasks(connectInfo.getTasks())
|
|
|
- .topics(ct.getTopics())
|
|
|
- .build()
|
|
|
- );
|
|
|
+ .map(ct -> connectInfo.toBuilder().topics(ct.getTopics()).build());
|
|
|
})
|
|
|
.map(kafkaConnectMapper::fullConnectorInfoFromTuple)
|
|
|
.filter(matchesSearchTerm(search));
|
|
@@ -132,6 +117,11 @@ public class KafkaConnectService {
|
|
|
.flatMapMany(Flux::fromIterable);
|
|
|
}
|
|
|
|
|
|
+ // returns empty flux if there was an error communicating with Connect
|
|
|
+ public Flux<String> getConnectorNamesWithErrorsSuppress(KafkaCluster cluster, String connectName) {
|
|
|
+ return getConnectorNames(cluster, connectName).onErrorComplete();
|
|
|
+ }
|
|
|
+
|
|
|
@SneakyThrows
|
|
|
private List<String> parseConnectorsNamesStringToList(String json) {
|
|
|
return objectMapper.readValue(json, new TypeReference<>() {
|