KC: Stop Connectors and Reset Connector Offsets

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

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

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

contract/pom.xml

@@ -201,6 +201,7 @@
  </goals>
  <configuration>
  <arguments>gen:sources</arguments>
+ <pnpmInheritsProxyConfigFromMaven>false</pnpmInheritsProxyConfigFromMaven>
  </configuration>
  </execution>
  </executions>

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'