KC: Stop Connectors and Reset Connector Offsets (#573)

Co-authored-by: NOZAIS Julien <[email protected]> Co-authored-by: Roman Zabaluev <[email protected]>
kafbat · Jan 14, 2025 · a159ef6 · a159ef6
1 parent aaac4d7
commit a159ef6
Show file tree

Hide file tree

Showing 19 changed files with 517 additions and 39 deletions.
diff --git a/.mvn/jvm.config b/.mvn/jvm.config
@@ -0,0 +1 @@
+-Djava.net.useSystemProxies=true
diff --git a/api/pom.xml b/api/pom.xml
@@ -492,6 +492,7 @@
                                 </goals>
                                 <configuration>
                                     <arguments>build</arguments>
+                                    <pnpmInheritsProxyConfigFromMaven>false</pnpmInheritsProxyConfigFromMaven>
                                 </configuration>
                             </execution>
                         </executions>

diff --git a/api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java b/api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java
@@ -238,6 +238,16 @@ public Mono<ResponseEntity<Void>> pauseConnectorWithHttpInfo(String connectorNam
     return withRetryOnConflictOrRebalance(super.pauseConnectorWithHttpInfo(connectorName));
   }
 
+  @Override
+  public Mono<Void> stopConnector(String connectorName) throws WebClientResponseException {
+    return withRetryOnConflictOrRebalance(super.stopConnector(connectorName));
+  }
+
+  @Override
+  public Mono<ResponseEntity<Void>> stopConnectorWithHttpInfo(String connectorName) throws WebClientResponseException {
+    return withRetryOnConflictOrRebalance(super.stopConnectorWithHttpInfo(connectorName));
+  }
+
   @Override
   public Mono<Void> restartConnector(String connectorName, Boolean includeTasks, Boolean onlyFailed)
       throws WebClientResponseException {
@@ -261,6 +271,18 @@ public Mono<ResponseEntity<Void>> restartConnectorTaskWithHttpInfo(String connec
     return withRetryOnConflictOrRebalance(super.restartConnectorTaskWithHttpInfo(connectorName, taskId));
   }
 
+  @Override
+  public Mono<Void> resetConnectorOffsets(String connectorName)
+      throws WebClientResponseException {
+    return withRetryOnConflictOrRebalance(super.resetConnectorOffsets(connectorName));
+  }
+
+  @Override
+  public Mono<ResponseEntity<Void>> resetConnectorOffsetsWithHttpInfo(String connectorName)
+      throws WebClientResponseException {
+    return withRetryOnConflictOrRebalance(super.resetConnectorOffsetsWithHttpInfo(connectorName));
+  }
+
   @Override
   public Mono<Void> resumeConnector(String connectorName) throws WebClientResponseException {
     return withRetryOnRebalance(super.resumeConnector(connectorName));

diff --git a/api/src/main/java/io/kafbat/ui/controller/KafkaConnectController.java b/api/src/main/java/io/kafbat/ui/controller/KafkaConnectController.java
@@ -3,6 +3,8 @@
 import static io.kafbat.ui.model.ConnectorActionDTO.RESTART;
 import static io.kafbat.ui.model.ConnectorActionDTO.RESTART_ALL_TASKS;
 import static io.kafbat.ui.model.ConnectorActionDTO.RESTART_FAILED_TASKS;
+import static io.kafbat.ui.model.rbac.permission.ConnectAction.RESET_OFFSETS;
+import static io.kafbat.ui.model.rbac.permission.ConnectAction.VIEW;
 
 import io.kafbat.ui.api.KafkaConnectApi;
 import io.kafbat.ui.model.ConnectDTO;
@@ -285,4 +287,23 @@ private Comparator<FullConnectorInfoDTO> getConnectorsComparator(ConnectorColumn
       default -> defaultComparator;
     };
   }
+
+  @Override
+  public Mono<ResponseEntity<Void>> resetConnectorOffsets(String clusterName, String connectName,
+      String connectorName,
+      ServerWebExchange exchange) {
+
+    var context = AccessContext.builder()
+        .cluster(clusterName)
+        .connectActions(connectName, VIEW, RESET_OFFSETS)
+        .operationName("resetConnectorOffsets")
+        .operationParams(Map.of(CONNECTOR_NAME, connectorName))
+        .build();
+
+    return validateAccess(context).then(
+        kafkaConnectService
+            .resetConnectorOffsets(getCluster(clusterName), connectName, connectorName)
+            .map(ResponseEntity::ok))
+        .doOnEach(sig -> audit(context, sig));
+  }
 }
diff --git a/api/src/main/java/io/kafbat/ui/exception/ConnectorOffsetsResetException.java b/api/src/main/java/io/kafbat/ui/exception/ConnectorOffsetsResetException.java
@@ -0,0 +1,13 @@
+package io.kafbat.ui.exception;
+
+public class ConnectorOffsetsResetException extends CustomBaseException {
+
+  public ConnectorOffsetsResetException(String message) {
+    super(message);
+  }
+
+  @Override
+  public ErrorCode getErrorCode() {
+    return ErrorCode.CONNECTOR_OFFSETS_RESET_ERROR;
+  }
+}
diff --git a/api/src/main/java/io/kafbat/ui/exception/ErrorCode.java b/api/src/main/java/io/kafbat/ui/exception/ErrorCode.java
@@ -32,6 +32,7 @@ public enum ErrorCode {
   TOPIC_ANALYSIS_ERROR(4018, HttpStatus.BAD_REQUEST),
   FILE_UPLOAD_EXCEPTION(4019, HttpStatus.INTERNAL_SERVER_ERROR),
   CEL_ERROR(4020, HttpStatus.BAD_REQUEST),
+  CONNECTOR_OFFSETS_RESET_ERROR(4021, HttpStatus.BAD_REQUEST),
   ;
 
   static {

diff --git a/api/src/main/java/io/kafbat/ui/model/rbac/permission/ConnectAction.java b/api/src/main/java/io/kafbat/ui/model/rbac/permission/ConnectAction.java
@@ -10,7 +10,8 @@ public enum ConnectAction implements PermissibleAction {
   EDIT(VIEW),
   CREATE(VIEW),
   RESTART(VIEW),
-  DELETE(VIEW)
+  DELETE(VIEW),
+  RESET_OFFSETS(VIEW)
 
   ;
 
@@ -20,7 +21,7 @@ public enum ConnectAction implements PermissibleAction {
     this.dependantActions = dependantActions;
   }
 
-  public static final Set<ConnectAction> ALTER_ACTIONS = Set.of(CREATE, EDIT, DELETE, RESTART);
+  public static final Set<ConnectAction> ALTER_ACTIONS = Set.of(CREATE, EDIT, DELETE, RESTART, RESET_OFFSETS);
 
   @Nullable
   public static ConnectAction fromString(String name) {

diff --git a/api/src/main/java/io/kafbat/ui/service/KafkaConnectService.java b/api/src/main/java/io/kafbat/ui/service/KafkaConnectService.java
@@ -6,6 +6,7 @@
 import io.kafbat.ui.connect.model.ConnectorStatusConnector;
 import io.kafbat.ui.connect.model.ConnectorTopics;
 import io.kafbat.ui.connect.model.TaskStatus;
+import io.kafbat.ui.exception.ConnectorOffsetsResetException;
 import io.kafbat.ui.exception.NotFoundException;
 import io.kafbat.ui.exception.ValidationException;
 import io.kafbat.ui.mapper.ClusterMapper;
@@ -213,6 +214,7 @@ public Mono<Void> updateConnectorState(KafkaCluster cluster, String connectName,
           case RESTART_FAILED_TASKS -> restartTasks(cluster, connectName, connectorName,
               t -> t.getStatus().getState() == ConnectorTaskStatusDTO.FAILED);
           case PAUSE -> client.pauseConnector(connectorName);
+          case STOP -> client.stopConnector(connectorName);
           case RESUME -> client.resumeConnector(connectorName);
         });
   }
@@ -272,4 +274,20 @@ private ReactiveFailover<KafkaConnectClientApi> api(KafkaCluster cluster, String
     }
     return client;
   }
+
+  public Mono<Void> resetConnectorOffsets(KafkaCluster cluster, String connectName,
+      String connectorName) {
+    return api(cluster, connectName)
+        .mono(client -> client.resetConnectorOffsets(connectorName))
+        .onErrorResume(WebClientResponseException.NotFound.class,
+            e -> {
+              throw new NotFoundException("Connector %s not found in %s".formatted(connectorName, connectName));
+            })
+        .onErrorResume(WebClientResponseException.BadRequest.class,
+            e -> {
+              throw new ConnectorOffsetsResetException(
+                  "Failed to reset offsets of connector %s of %s. Make sure it is STOPPED first."
+                      .formatted(connectorName, connectName));
+            });
+  }
 }
diff --git a/api/src/test/java/io/kafbat/ui/KafkaConnectServiceTests.java b/api/src/test/java/io/kafbat/ui/KafkaConnectServiceTests.java
@@ -24,6 +24,8 @@
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.HttpStatus;
+import org.springframework.test.web.reactive.server.ExchangeResult;
 import org.springframework.test.web.reactive.server.WebTestClient;
 
 @Slf4j
@@ -45,6 +47,7 @@ public class KafkaConnectServiceTests extends AbstractIntegrationTest {
 
   @BeforeEach
   public void setUp() {
+
     webTestClient.post()
         .uri("/api/clusters/{clusterName}/connects/{connectName}/connectors", LOCAL, connectName)
         .bodyValue(new NewConnectorDTO()
@@ -54,11 +57,10 @@ public void setUp() {
                 "tasks.max", "1",
                 "topics", "output-topic",
                 "file", "/tmp/test",
-                "test.password", "test-credentials"
-            ))
-        )
+                "test.password", "test-credentials")))
         .exchange()
         .expectStatus().isOk();
+
   }
 
   @AfterEach
@@ -418,4 +420,56 @@ public void shouldReturn400WhenTryingToCreateConnectorWithExistingName() {
         .expectStatus()
         .isBadRequest();
   }
+
+  @Test
+  public void shouldResetConnectorWhenInStoppedState() {
+
+    webTestClient.get()
+        .uri("/api/clusters/{clusterName}/connects/{connectName}/connectors/{connectorName}",
+            LOCAL, connectName, connectorName)
+        .exchange()
+        .expectStatus().isOk()
+        .expectBody(ConnectorDTO.class)
+        .value(connector -> assertThat(connector.getStatus().getState()).isEqualTo(ConnectorStateDTO.RUNNING));
+
+    webTestClient.post()
+        .uri("/api/clusters/{clusterName}/connects/{connectName}/connectors/{connectorName}/action/STOP",
+            LOCAL, connectName, connectorName)
+        .exchange()
+        .expectStatus().isOk();
+
+    webTestClient.get()
+        .uri("/api/clusters/{clusterName}/connects/{connectName}/connectors/{connectorName}",
+            LOCAL, connectName, connectorName)
+        .exchange()
+        .expectStatus().isOk()
+        .expectBody(ConnectorDTO.class)
+        .value(connector -> assertThat(connector.getStatus().getState()).isEqualTo(ConnectorStateDTO.STOPPED));
+
+    webTestClient.delete()
+        .uri("/api/clusters/{clusterName}/connects/{connectName}/connectors/{connectorName}/offsets",
+            LOCAL, connectName, connectorName)
+        .exchange()
+        .expectStatus().isOk();
+
+  }
+
+  @Test
+  public void shouldReturn400WhenResettingConnectorInRunningState() {
+
+    webTestClient.get()
+        .uri("/api/clusters/{clusterName}/connects/{connectName}/connectors/{connectorName}",
+            LOCAL, connectName, connectorName)
+        .exchange()
+        .expectStatus().isOk()
+        .expectBody(ConnectorDTO.class)
+        .value(connector -> assertThat(connector.getStatus().getState()).isEqualTo(ConnectorStateDTO.RUNNING));
+
+    webTestClient.delete()
+        .uri("/api/clusters/{clusterName}/connects/{connectName}/connectors/{connectorName}/offsets", LOCAL,
+            connectName, connectorName)
+        .exchange()
+        .expectStatus().isBadRequest();
+
+  }
 }
diff --git a/contract/pom.xml b/contract/pom.xml
@@ -201,6 +201,7 @@
                                 </goals>
                                 <configuration>
                                     <arguments>gen:sources</arguments>
+                                    <pnpmInheritsProxyConfigFromMaven>false</pnpmInheritsProxyConfigFromMaven>
                                 </configuration>
                             </execution>
                         </executions>

diff --git a/contract/src/main/resources/swagger/kafbat-ui-api.yaml b/contract/src/main/resources/swagger/kafbat-ui-api.yaml
@@ -1565,7 +1565,7 @@ paths:
     post:
       tags:
         - Kafka Connect
-      summary: update connector state (restart, pause or resume)
+      summary: update connector state (restart, pause, stop or resume)
       operationId: updateConnectorState
       parameters:
         - name: clusterName
@@ -1722,6 +1722,31 @@ paths:
         200:
           description: OK
 
+  /api/clusters/{clusterName}/connects/{connectName}/connectors/{connectorName}/offsets:
+    delete:
+      tags:
+        - Kafka Connect
+      summary: reset the offsets for the specified connector
+      operationId: resetConnectorOffsets
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: connectName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: connectorName
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        200:
+          description: OK
 
   /api/clusters/{clusterName}/ksql/v2:
     post:
@@ -3567,6 +3592,7 @@ components:
         - RESTART_FAILED_TASKS
         - PAUSE
         - RESUME
+        - STOP
 
     TaskAction:
       type: string
@@ -3953,7 +3979,16 @@ components:
 
     KafkaAcl:
       type: object
-      required: [resourceType, resourceName, namePatternType, principal, host, operation, permission]
+      required:
+        [
+          resourceType,
+          resourceName,
+          namePatternType,
+          principal,
+          host,
+          operation,
+          permission,
+        ]
       properties:
         resourceType:
           $ref: '#/components/schemas/KafkaAclResourceType'