KC: Stop Connectors and Reset Connector Offsets

.mvn/jvm.config

@@ -0,0 +1 @@
+-Djava.net.useSystemProxies=true

api/pom.xml

@@ -502,6 +502,7 @@
                                 </goals>
                                 <configuration>
                                     <arguments>build</arguments>
+                                    <pnpmInheritsProxyConfigFromMaven>false</pnpmInheritsProxyConfigFromMaven>
                                 </configuration>
                             </execution>
                         </executions>

api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java

@@ -190,6 +190,11 @@ public Mono<Void> pauseConnector(String connectorName) throws WebClientResponseE
     return withRetryOnConflict(super.pauseConnector(connectorName));
   }
 
+  @Override
+  public Mono<Void> stopConnector(String connectorName) throws WebClientResponseException {
+    return withRetryOnConflict(super.stopConnector(connectorName));
+  }
+
   @Override
   public Mono<ResponseEntity<Void>> pauseConnectorWithHttpInfo(String connectorName) throws WebClientResponseException {
     return withRetryOnConflict(super.pauseConnectorWithHttpInfo(connectorName));

api/src/main/java/io/kafbat/ui/controller/KafkaConnectController.java

@@ -34,15 +34,15 @@
 @RequiredArgsConstructor
 @Slf4j
 public class KafkaConnectController extends AbstractController implements KafkaConnectApi {
-  private static final Set<ConnectorActionDTO> RESTART_ACTIONS
-      = Set.of(RESTART, RESTART_FAILED_TASKS, RESTART_ALL_TASKS);
+  private static final Set<ConnectorActionDTO> RESTART_ACTIONS = Set.of(RESTART, RESTART_FAILED_TASKS,
+      RESTART_ALL_TASKS);
   private static final String CONNECTOR_NAME = "connectorName";
 
   private final KafkaConnectService kafkaConnectService;
 
   @Override
   public Mono<ResponseEntity<Flux<ConnectDTO>>> getConnects(String clusterName,
-                                                            ServerWebExchange exchange) {
+      ServerWebExchange exchange) {
 
     Flux<ConnectDTO> availableConnects = kafkaConnectService.getConnects(getCluster(clusterName))
         .filterWhen(dto -> accessControlService.isConnectAccessible(dto, clusterName));
@@ -52,7 +52,7 @@ public Mono<ResponseEntity<Flux<ConnectDTO>>> getConnects(String clusterName,
 
   @Override
   public Mono<ResponseEntity<Flux<String>>> getConnectors(String clusterName, String connectName,
-                                                          ServerWebExchange exchange) {
+      ServerWebExchange exchange) {
 
     var context = AccessContext.builder()
         .cluster(clusterName)
@@ -61,14 +61,15 @@ public Mono<ResponseEntity<Flux<String>>> getConnectors(String clusterName, Stri
         .build();
 
     return validateAccess(context)
-        .thenReturn(ResponseEntity.ok(kafkaConnectService.getConnectorNames(getCluster(clusterName), connectName)))
+        .thenReturn(
+            ResponseEntity.ok(kafkaConnectService.getConnectorNames(getCluster(clusterName), connectName)))
         .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
   public Mono<ResponseEntity<ConnectorDTO>> createConnector(String clusterName, String connectName,
-                                                            @Valid Mono<NewConnectorDTO> connector,
-                                                            ServerWebExchange exchange) {
+      @Valid Mono<NewConnectorDTO> connector,
+      ServerWebExchange exchange) {
 
     var context = AccessContext.builder()
         .cluster(clusterName)
@@ -78,14 +79,14 @@ public Mono<ResponseEntity<ConnectorDTO>> createConnector(String clusterName, St
 
     return validateAccess(context).then(
         kafkaConnectService.createConnector(getCluster(clusterName), connectName, connector)
-            .map(ResponseEntity::ok)
-    ).doOnEach(sig -> audit(context, sig));
+            .map(ResponseEntity::ok))
+        .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
   public Mono<ResponseEntity<ConnectorDTO>> getConnector(String clusterName, String connectName,
-                                                         String connectorName,
-                                                         ServerWebExchange exchange) {
+      String connectorName,
+      ServerWebExchange exchange) {
 
     var context = AccessContext.builder()
         .cluster(clusterName)
@@ -95,14 +96,14 @@ public Mono<ResponseEntity<ConnectorDTO>> getConnector(String clusterName, Strin
 
     return validateAccess(context).then(
         kafkaConnectService.getConnector(getCluster(clusterName), connectName, connectorName)
-            .map(ResponseEntity::ok)
-    ).doOnEach(sig -> audit(context, sig));
+            .map(ResponseEntity::ok))
+        .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
   public Mono<ResponseEntity<Void>> deleteConnector(String clusterName, String connectName,
-                                                    String connectorName,
-                                                    ServerWebExchange exchange) {
+      String connectorName,
+      ServerWebExchange exchange) {
 
     var context = AccessContext.builder()
         .cluster(clusterName)
@@ -113,19 +114,17 @@ public Mono<ResponseEntity<Void>> deleteConnector(String clusterName, String con
 
     return validateAccess(context).then(
         kafkaConnectService.deleteConnector(getCluster(clusterName), connectName, connectorName)
-            .map(ResponseEntity::ok)
-    ).doOnEach(sig -> audit(context, sig));
+            .map(ResponseEntity::ok))
+        .doOnEach(sig -> audit(context, sig));
   }
 
-
   @Override
   public Mono<ResponseEntity<Flux<FullConnectorInfoDTO>>> getAllConnectors(
       String clusterName,
       String search,
       ConnectorColumnsToSortDTO orderBy,
       SortOrderDTO sortOrder,
-      ServerWebExchange exchange
-  ) {
+      ServerWebExchange exchange) {
     var context = AccessContext.builder()
         .cluster(clusterName)
         .operationName("getAllConnectors")
@@ -145,9 +144,9 @@ public Mono<ResponseEntity<Flux<FullConnectorInfoDTO>>> getAllConnectors(
 
   @Override
   public Mono<ResponseEntity<Map<String, Object>>> getConnectorConfig(String clusterName,
-                                                                      String connectName,
-                                                                      String connectorName,
-                                                                      ServerWebExchange exchange) {
+      String connectName,
+      String connectorName,
+      ServerWebExchange exchange) {
 
     var context = AccessContext.builder()
         .cluster(clusterName)
@@ -158,15 +157,15 @@ public Mono<ResponseEntity<Map<String, Object>>> getConnectorConfig(String clust
     return validateAccess(context).then(
         kafkaConnectService
             .getConnectorConfig(getCluster(clusterName), connectName, connectorName)
-            .map(ResponseEntity::ok)
-    ).doOnEach(sig -> audit(context, sig));
+            .map(ResponseEntity::ok))
+        .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
   public Mono<ResponseEntity<ConnectorDTO>> setConnectorConfig(String clusterName, String connectName,
-                                                               String connectorName,
-                                                               Mono<Map<String, Object>> requestBody,
-                                                               ServerWebExchange exchange) {
+      String connectorName,
+      Mono<Map<String, Object>> requestBody,
+      ServerWebExchange exchange) {
 
     var context = AccessContext.builder()
         .cluster(clusterName)
@@ -176,22 +175,22 @@ public Mono<ResponseEntity<ConnectorDTO>> setConnectorConfig(String clusterName,
         .build();
 
     return validateAccess(context).then(
-            kafkaConnectService
-                .setConnectorConfig(getCluster(clusterName), connectName, connectorName, requestBody)
-                .map(ResponseEntity::ok))
+        kafkaConnectService
+            .setConnectorConfig(getCluster(clusterName), connectName, connectorName, requestBody)
+            .map(ResponseEntity::ok))
         .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
   public Mono<ResponseEntity<Void>> updateConnectorState(String clusterName, String connectName,
-                                                         String connectorName,
-                                                         ConnectorActionDTO action,
-                                                         ServerWebExchange exchange) {
+      String connectorName,
+      ConnectorActionDTO action,
+      ServerWebExchange exchange) {
     ConnectAction[] connectActions;
     if (RESTART_ACTIONS.contains(action)) {
-      connectActions = new ConnectAction[] {ConnectAction.VIEW, ConnectAction.RESTART};
+      connectActions = new ConnectAction[] { ConnectAction.VIEW, ConnectAction.RESTART };
     } else {
-      connectActions = new ConnectAction[] {ConnectAction.VIEW, ConnectAction.EDIT};
+      connectActions = new ConnectAction[] { ConnectAction.VIEW, ConnectAction.EDIT };
     }
 
     var context = AccessContext.builder()
@@ -204,15 +203,15 @@ public Mono<ResponseEntity<Void>> updateConnectorState(String clusterName, Strin
     return validateAccess(context).then(
         kafkaConnectService
             .updateConnectorState(getCluster(clusterName), connectName, connectorName, action)
-            .map(ResponseEntity::ok)
-    ).doOnEach(sig -> audit(context, sig));
+            .map(ResponseEntity::ok))
+        .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
   public Mono<ResponseEntity<Flux<TaskDTO>>> getConnectorTasks(String clusterName,
-                                                               String connectName,
-                                                               String connectorName,
-                                                               ServerWebExchange exchange) {
+      String connectName,
+      String connectorName,
+      ServerWebExchange exchange) {
     var context = AccessContext.builder()
         .cluster(clusterName)
         .connectActions(connectName, ConnectAction.VIEW)
@@ -223,14 +222,14 @@ public Mono<ResponseEntity<Flux<TaskDTO>>> getConnectorTasks(String clusterName,
     return validateAccess(context).thenReturn(
         ResponseEntity
             .ok(kafkaConnectService
-                .getConnectorTasks(getCluster(clusterName), connectName, connectorName))
-    ).doOnEach(sig -> audit(context, sig));
+                .getConnectorTasks(getCluster(clusterName), connectName, connectorName)))
+        .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
   public Mono<ResponseEntity<Void>> restartConnectorTask(String clusterName, String connectName,
-                                                         String connectorName, Integer taskId,
-                                                         ServerWebExchange exchange) {
+      String connectorName, Integer taskId,
+      ServerWebExchange exchange) {
 
     var context = AccessContext.builder()
         .cluster(clusterName)
@@ -242,8 +241,8 @@ public Mono<ResponseEntity<Void>> restartConnectorTask(String clusterName, Strin
     return validateAccess(context).then(
         kafkaConnectService
             .restartConnectorTask(getCluster(clusterName), connectName, connectorName, taskId)
-            .map(ResponseEntity::ok)
-    ).doOnEach(sig -> audit(context, sig));
+            .map(ResponseEntity::ok))
+        .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
@@ -259,8 +258,8 @@ public Mono<ResponseEntity<Flux<ConnectorPluginDTO>>> getConnectorPlugins(
     return validateAccess(context).then(
         Mono.just(
             ResponseEntity.ok(
-                kafkaConnectService.getConnectorPlugins(getCluster(clusterName), connectName)))
-    ).doOnEach(sig -> audit(context, sig));
+                kafkaConnectService.getConnectorPlugins(getCluster(clusterName), connectName))))
+        .doOnEach(sig -> audit(context, sig));
   }
 
   @Override
@@ -285,4 +284,26 @@ private Comparator<FullConnectorInfoDTO> getConnectorsComparator(ConnectorColumn
       default -> defaultComparator;
     };
   }
+
+  @Override
+  public Mono<ResponseEntity<Void>> resetConnectorOffsets(String clusterName, String connectName,
+      String connectorName,
+      ServerWebExchange exchange) {
+    ConnectAction[] connectActions;
+
+    connectActions = new ConnectAction[] { ConnectAction.VIEW, ConnectAction.RESET_OFFSETS };
+
+    var context = AccessContext.builder()
+        .cluster(clusterName)
+        .connectActions(connectName, connectActions)
+        .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));
+  }
 }

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) {