Repository: provectus/kafka-ui
Issue: Connector "Restart All Tasks" button has no effect
Labels: type/enhancement
scope/backend
scope/frontend
status/accepted
status/confirmed

When choosing a specific connector there are several options you can choose:
<img width="1083" alt="image" src="https://user-images.githubusercontent.com/94184844/150494050-67a4692f-2133-4c78-94e3-c74857b65057.png">

Restart all tasks sends the next POST request that actually does do nothing.
https://xxx/api/clusters/local/connects/local/connectors/reddit-source-json/action/RESTART

According to the REST API documentation https://docs.confluent.io/platform/current/connect/references/restapi.html there is no such request available. And the correct one is this:
<img width="904" alt="image" src="https://user-images.githubusercontent.com/94184844/150494299-2cffab5d-abb7-4e05-af35-f3af1a57a050.png">

Restarting the specific task using the triple dot context menu works great tho.

Issue comments:
@5hin0bi we currently only restart **connector instance**, not tasks on this button click. I will change this to restart both connector and all its tasks. I think we will also need to rename this button to smth like "Restart connector". 
cc @Haarolean 
@5hin0bi can you please verify it works, and reopen issue if not
Checked on our dev environment. The button is now called "Restart Connector", but failed tasks themselves are not restarted.
To do: create 3 buttons
Need to implement frontend
I'm seeing a similar error with KafkaConnect where it's trying to hit an api endpoint:
```500 Server Error for HTTP GET "/api/clusters/local/connectors?search="```
that also doesn't seem to exist, per Confluent's API. 
> I'm seeing a similar error with KafkaConnect where it's trying to hit an api endpoint: `500 Server Error for HTTP GET "/api/clusters/local/connectors?search="` that also doesn't seem to exist, per Confluent's API.

That's our backend with our API, not confluent's. Please raise a new issue.

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java
index 2d82a54217b..1c3d2474924 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java
@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.provectus.kafka.ui.client.KafkaConnectClients;
+import com.provectus.kafka.ui.connect.api.KafkaConnectClientApi;
 import com.provectus.kafka.ui.connect.model.ConnectorStatus;
 import com.provectus.kafka.ui.connect.model.ConnectorStatusConnector;
 import com.provectus.kafka.ui.connect.model.ConnectorTopics;
@@ -17,6 +18,7 @@
 import com.provectus.kafka.ui.model.ConnectorPluginConfigValidationResponseDTO;
 import com.provectus.kafka.ui.model.ConnectorPluginDTO;
 import com.provectus.kafka.ui.model.ConnectorStateDTO;
+import com.provectus.kafka.ui.model.ConnectorTaskStatusDTO;
 import com.provectus.kafka.ui.model.FullConnectorInfoDTO;
 import com.provectus.kafka.ui.model.KafkaCluster;
 import com.provectus.kafka.ui.model.KafkaConnectCluster;
@@ -119,12 +121,11 @@ private Stream<String> getSearchValues(FullConnectorInfoDTO fullConnectorInfo) {
 
   private Mono<ConnectorTopics> getConnectorTopics(KafkaCluster cluster, String connectClusterName,
                                                    String connectorName) {
-    return getConnectAddress(cluster, connectClusterName)
-        .flatMap(connectUrl -> KafkaConnectClients
-            .withBaseUrl(connectUrl)
-            .getConnectorTopics(connectorName)
-            .map(result -> result.get(connectorName))
-        );
+    return withConnectClient(cluster, connectClusterName)
+        .flatMap(c -> c.getConnectorTopics(connectorName).map(result -> result.get(connectorName)))
+        // old connectors don't have this api, setting empty list for
+        // backward-compatibility
+        .onErrorResume(Exception.class, e -> Mono.just(new ConnectorTopics().topics(List.of())));
   }
 
   private Flux<Tuple2<String, String>> getConnectorNames(KafkaCluster cluster, String connectName) {
@@ -143,17 +144,17 @@ private List<String> parseToList(String json) {
   }
 
   public Flux<String> getConnectors(KafkaCluster cluster, String connectName) {
-    return getConnectAddress(cluster, connectName)
-        .flatMapMany(connect ->
-            KafkaConnectClients.withBaseUrl(connect).getConnectors(null)
+    return withConnectClient(cluster, connectName)
+        .flatMapMany(client ->
+            client.getConnectors(null)
                 .doOnError(e -> log.error("Unexpected error upon getting connectors", e))
         );
   }
 
   public Mono<ConnectorDTO> createConnector(KafkaCluster cluster, String connectName,
                                             Mono<NewConnectorDTO> connector) {
-    return getConnectAddress(cluster, connectName)
-        .flatMap(connectUrl ->
+    return withConnectClient(cluster, connectName)
+        .flatMap(client ->
             connector
                 .flatMap(c -> connectorExists(cluster, connectName, c.getName())
                     .map(exists -> {
@@ -164,7 +165,7 @@ public Mono<ConnectorDTO> createConnector(KafkaCluster cluster, String connectNa
                       return c;
                     }))
                 .map(kafkaConnectMapper::toClient)
-                .flatMap(c -> KafkaConnectClients.withBaseUrl(connectUrl).createConnector(c))
+                .flatMap(client::createConnector)
                 .flatMap(c -> getConnector(cluster, connectName, c.getName()))
         );
   }
@@ -179,11 +180,11 @@ private Mono<Boolean> connectorExists(KafkaCluster cluster, String connectName,
 
   public Mono<ConnectorDTO> getConnector(KafkaCluster cluster, String connectName,
                                       String connectorName) {
-    return getConnectAddress(cluster, connectName)
-        .flatMap(connect -> KafkaConnectClients.withBaseUrl(connect).getConnector(connectorName)
+    return withConnectClient(cluster, connectName)
+        .flatMap(client -> client.getConnector(connectorName)
             .map(kafkaConnectMapper::fromClient)
             .flatMap(connector ->
-                KafkaConnectClients.withBaseUrl(connect).getConnectorStatus(connector.getName())
+                client.getConnectorStatus(connector.getName())
                     // status request can return 404 if tasks not assigned yet
                     .onErrorResume(WebClientResponseException.NotFound.class,
                         e -> emptyStatus(connectorName))
@@ -228,10 +229,8 @@ private Mono<ConnectorStatus> emptyStatus(String connectorName) {
 
   public Mono<Map<String, Object>> getConnectorConfig(KafkaCluster cluster, String connectName,
                                                       String connectorName) {
-    return getConnectAddress(cluster, connectName)
-        .flatMap(connect ->
-            KafkaConnectClients.withBaseUrl(connect).getConnectorConfig(connectorName)
-        )
+    return withConnectClient(cluster, connectName)
+        .flatMap(c -> c.getConnectorConfig(connectorName))
         .map(connectorConfig -> {
           final Map<String, Object> obfuscatedMap = new HashMap<>();
           connectorConfig.forEach((key, value) ->
@@ -242,99 +241,94 @@ public Mono<Map<String, Object>> getConnectorConfig(KafkaCluster cluster, String
 
   public Mono<ConnectorDTO> setConnectorConfig(KafkaCluster cluster, String connectName,
                                             String connectorName, Mono<Object> requestBody) {
-    return getConnectAddress(cluster, connectName)
-        .flatMap(connect ->
-            requestBody.flatMap(body ->
-                KafkaConnectClients.withBaseUrl(connect)
-                    .setConnectorConfig(connectorName, (Map<String, Object>) body)
-            )
-                .map(kafkaConnectMapper::fromClient)
-        );
+    return  withConnectClient(cluster, connectName)
+        .flatMap(c ->
+            requestBody
+                .flatMap(body -> c.setConnectorConfig(connectorName, (Map<String, Object>) body))
+                .map(kafkaConnectMapper::fromClient));
   }
 
   public Mono<Void> deleteConnector(
       KafkaCluster cluster, String connectName, String connectorName) {
-    return getConnectAddress(cluster, connectName)
-        .flatMap(connect ->
-            KafkaConnectClients.withBaseUrl(connect).deleteConnector(connectorName)
-        );
+    return withConnectClient(cluster, connectName)
+        .flatMap(c -> c.deleteConnector(connectorName));
   }
 
   public Mono<Void> updateConnectorState(KafkaCluster cluster, String connectName,
                                          String connectorName, ConnectorActionDTO action) {
-    Function<String, Mono<Void>> kafkaClientCall;
-    switch (action) {
-      case RESTART:
-        kafkaClientCall =
-            connect -> KafkaConnectClients.withBaseUrl(connect)
-                .restartConnector(connectorName, true, false);
-        break;
-      case PAUSE:
-        kafkaClientCall =
-            connect -> KafkaConnectClients.withBaseUrl(connect).pauseConnector(connectorName);
-        break;
-      case RESUME:
-        kafkaClientCall =
-            connect -> KafkaConnectClients.withBaseUrl(connect).resumeConnector(connectorName);
-        break;
-      default:
-        throw new IllegalStateException("Unexpected value: " + action);
-    }
-    return getConnectAddress(cluster, connectName)
-        .flatMap(kafkaClientCall);
+    return withConnectClient(cluster, connectName)
+        .flatMap(client -> {
+          switch (action) {
+            case RESTART:
+              return client.restartConnector(connectorName, false, false);
+            case RESTART_ALL_TASKS:
+              return restartTasks(cluster, connectName, connectorName, task -> true);
+            case RESTART_FAILED_TASKS:
+              return restartTasks(cluster, connectName, connectorName,
+                  t -> t.getStatus().getState() == ConnectorTaskStatusDTO.FAILED);
+            case PAUSE:
+              return client.pauseConnector(connectorName);
+            case RESUME:
+              return client.resumeConnector(connectorName);
+            default:
+              throw new IllegalStateException("Unexpected value: " + action);
+          }
+        });
   }
 
-  public Flux<TaskDTO> getConnectorTasks(KafkaCluster cluster, String connectName,
-                                      String connectorName) {
-    return getConnectAddress(cluster, connectName)
-        .flatMapMany(connect ->
-            KafkaConnectClients.withBaseUrl(connect).getConnectorTasks(connectorName)
+  private Mono<Void> restartTasks(KafkaCluster cluster, String connectName,
+                                  String connectorName, Predicate<TaskDTO> taskFilter) {
+    return getConnectorTasks(cluster, connectName, connectorName)
+        .filter(taskFilter)
+        .flatMap(t ->
+            restartConnectorTask(cluster, connectName, connectorName, t.getId().getTask()))
+        .then();
+  }
+
+  public Flux<TaskDTO> getConnectorTasks(KafkaCluster cluster, String connectName, String connectorName) {
+    return withConnectClient(cluster, connectName)
+        .flatMapMany(client ->
+            client.getConnectorTasks(connectorName)
                 .onErrorResume(WebClientResponseException.NotFound.class, e -> Flux.empty())
                 .map(kafkaConnectMapper::fromClient)
                 .flatMap(task ->
-                    KafkaConnectClients.withBaseUrl(connect)
+                    client
                         .getConnectorTaskStatus(connectorName, task.getId().getTask())
                         .onErrorResume(WebClientResponseException.NotFound.class, e -> Mono.empty())
                         .map(kafkaConnectMapper::fromClient)
                         .map(task::status)
-                )
-        );
+                ));
   }
 
   public Mono<Void> restartConnectorTask(KafkaCluster cluster, String connectName,
                                          String connectorName, Integer taskId) {
-    return getConnectAddress(cluster, connectName)
-        .flatMap(connect ->
-            KafkaConnectClients.withBaseUrl(connect).restartConnectorTask(connectorName, taskId)
-        );
+    return withConnectClient(cluster, connectName)
+        .flatMap(client -> client.restartConnectorTask(connectorName, taskId));
   }
 
   public Mono<Flux<ConnectorPluginDTO>> getConnectorPlugins(KafkaCluster cluster,
                                                             String connectName) {
-    return Mono.just(getConnectAddress(cluster, connectName)
-        .flatMapMany(connect ->
-            KafkaConnectClients.withBaseUrl(connect).getConnectorPlugins()
-                .map(kafkaConnectMapper::fromClient)
-        ));
+    return withConnectClient(cluster, connectName)
+        .map(client -> client.getConnectorPlugins().map(kafkaConnectMapper::fromClient));
   }
 
   public Mono<ConnectorPluginConfigValidationResponseDTO> validateConnectorPluginConfig(
       KafkaCluster cluster, String connectName, String pluginName, Mono<Object> requestBody) {
-    return getConnectAddress(cluster, connectName)
-        .flatMap(connect ->
-            requestBody.flatMap(body ->
-                KafkaConnectClients.withBaseUrl(connect)
-                    .validateConnectorPluginConfig(pluginName, (Map<String, Object>) body)
-            )
+    return withConnectClient(cluster, connectName)
+        .flatMap(client ->
+            requestBody
+                .flatMap(body ->
+                    client.validateConnectorPluginConfig(pluginName, (Map<String, Object>) body))
                 .map(kafkaConnectMapper::fromClient)
         );
   }
 
-  private Mono<String> getConnectAddress(KafkaCluster cluster, String connectName) {
+  private Mono<KafkaConnectClientApi> withConnectClient(KafkaCluster cluster, String connectName) {
     return Mono.justOrEmpty(cluster.getKafkaConnect().stream()
             .filter(connect -> connect.getName().equals(connectName))
             .findFirst()
             .map(KafkaConnectCluster::getAddress))
-        .switchIfEmpty(Mono.error(ConnectNotFoundException::new));
+        .switchIfEmpty(Mono.error(ConnectNotFoundException::new))
+        .map(KafkaConnectClients::withBaseUrl);
   }
 }
diff --git a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
index 277c6d71520..318366ed44f 100644
--- a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
+++ b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
@@ -2492,6 +2492,8 @@ components:
       type: string
       enum:
         - RESTART
+        - RESTART_ALL_TASKS
+        - RESTART_FAILED_TASKS
         - PAUSE
         - RESUME
 


Test patch: