Merge branch 'main' into pnpm-9.15.4

Author: yeikel

.github/workflows/frontend_tests.yml

@@ -28,7 +28,7 @@ jobs:
       - name: Install node
         uses: actions/[email protected]
         with:
-          node-version: "18.17.1"
+          node-version: "22.12.0"
           cache: "pnpm"
           cache-dependency-path: "./frontend/pnpm-lock.yaml"
 

.mvn/jvm.config

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

api/pom.xml

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

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

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));
+  }
 }

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;
+  }
+}

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 {

api/src/main/java/io/kafbat/ui/model/BrokerMetrics.java

@@ -1,6 +1,5 @@
 package io.kafbat.ui.model;
 
-import io.kafbat.ui.model.MetricDTO;
 import java.util.List;
 import lombok.Builder;
 import lombok.Data;

api/src/main/java/io/kafbat/ui/model/CleanupPolicy.java

@@ -20,18 +20,10 @@ public enum CleanupPolicy {
     this.policies = policies;
   }
 
-  public String getPolicy() {
-    return policies.get(0);
-  }
-
   public static CleanupPolicy fromString(String string) {
     return Arrays.stream(CleanupPolicy.values())
-        .filter(v ->
-            v.policies.stream().anyMatch(
-                s -> s.equals(string.replace(" ", "")
-                )
-            )
-        ).findFirst()
+        .filter(v -> v.policies.stream().anyMatch(s -> s.equals(string.replace(" ", ""))))
+        .findFirst()
         .orElse(UNKNOWN);
   }
 }

api/src/main/java/io/kafbat/ui/model/InternalBrokerConfig.java

@@ -1,6 +1,5 @@
 package io.kafbat.ui.model;
 
-
 import java.util.List;
 import lombok.Builder;
 import lombok.Data;
@@ -17,13 +16,13 @@ public class InternalBrokerConfig {
   private final List<ConfigEntry.ConfigSynonym> synonyms;
 
   public static InternalBrokerConfig from(ConfigEntry configEntry, boolean readOnlyCluster) {
-    InternalBrokerConfig.InternalBrokerConfigBuilder builder = InternalBrokerConfig.builder()
+    return InternalBrokerConfig.builder()
         .name(configEntry.name())
         .value(configEntry.value())
         .source(configEntry.source())
         .isReadOnly(readOnlyCluster || configEntry.isReadOnly())
         .isSensitive(configEntry.isSensitive())
-        .synonyms(configEntry.synonyms());
-    return builder.build();
+        .synonyms(configEntry.synonyms())
+        .build();
   }
 }