Add support for altering and partially resetting connector offsets

README.md

@@ -554,6 +554,48 @@ kafkactl connector restart myConnector
 kafkactl connector stop myConnector
 ```
 
+#### Alter Offsets
+
+The `alter-offsets` command allows you to alter or partially reset the offsets of a connector.
+
+```console
+Usage: kafkactl connector alter-offsets [-hv] [-c=<optionalContext>] -f=<file> [-n=<optionalNamespace>]
+                                        <connector>
+
+Description: Alter or partially reset connector offsets.
+
+Parameters:
+      <connector>   Connector name.
+
+Options:
+  -c, --context=<optionalContext>
+                        Override context defined in config.
+  -f, --file=<file>     YAML file containing the connector offsets payload.
+  -h, --help            Show this help message and exit.
+  -n, --namespace=<optionalNamespace>
+                        Override namespace defined in config or YAML resources.
+  -v, --verbose         Enable the verbose mode.
+```
+
+The payload uses the Kafka Connect alter offsets format. Set `offset` to `null` to reset only the selected partition, or provide an offset map to alter it.
+
+```yaml
+offsets:
+  - partition:
+      filename: /data/input.txt
+    offset:
+      position: 100
+  - partition:
+      filename: /data/other.txt
+    offset: null
+```
+
+Example:
+
+```console
+kafkactl connector alter-offsets myConnector -f offsets.yaml
+```
+
 #### Reset Offsets
 
 The `reset-offsets` command allows you to fully resets the offsets of a connector.

src/main/java/com/michelin/kafkactl/client/NamespacedResourceClient.java

@@ -19,13 +19,15 @@
 package com.michelin.kafkactl.client;
 
 import com.michelin.kafkactl.model.Resource;
+import com.michelin.kafkactl.model.request.ConnectorOffsets;
 import com.michelin.kafkactl.model.request.DeleteResourceRequest;
 import io.micronaut.core.annotation.Nullable;
 import io.micronaut.http.HttpResponse;
 import io.micronaut.http.annotation.Body;
 import io.micronaut.http.annotation.Delete;
 import io.micronaut.http.annotation.Get;
 import io.micronaut.http.annotation.Header;
+import io.micronaut.http.annotation.Patch;
 import io.micronaut.http.annotation.Post;
 import io.micronaut.http.annotation.QueryValue;
 import io.micronaut.http.annotation.RequestBean;
@@ -253,6 +255,24 @@ HttpResponse<Resource> updateSubjectConfig(
     HttpResponse<Resource> resetConnectorOffsets(
             String namespace, String connector, @Header("Authorization") String token);
 
+    /**
+     * Alter offsets for a given connector.
+     *
+     * @param namespace The namespace
+     * @param connector The connector to alter offsets for
+     * @param offsets The offsets payload
+     * @param token The auth token
+     * @return The alter offsets response
+     */
+    @Patch("{namespace}/connectors/{connector}/offsets")
+    @Retryable(
+            delay = "${kafkactl.retry.delay}",
+            attempts = "${kafkactl.retry.attempt}",
+            multiplier = "${kafkactl.retry.multiplier}",
+            includes = ReadTimeoutException.class)
+    HttpResponse<Resource> alterConnectorOffsets(
+            String namespace, String connector, @Body ConnectorOffsets offsets, @Header("Authorization") String token);
+
     /**
      * List all available connect clusters for vaulting.
      *

src/main/java/com/michelin/kafkactl/command/connector/Connector.java

@@ -42,7 +42,7 @@
 /** Connectors subcommand. */
 @Command(
         name = "connector",
-        subcommands = {ConnectorResetOffsets.class},
+        subcommands = {ConnectorAlterOffsets.class, ConnectorResetOffsets.class},
         headerHeading = "@|bold Usage|@:",
         synopsisHeading = " ",
         descriptionHeading = "%n@|bold Description|@: ",

src/main/java/com/michelin/kafkactl/command/connector/ConnectorAlterOffsets.java

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package com.michelin.kafkactl.command.connector;
+
+import static com.michelin.kafkactl.model.Output.TABLE;
+import static com.michelin.kafkactl.util.constant.ResourceKind.CONNECTOR_RESET_OFFSETS_RESPONSE;
+
+import com.michelin.kafkactl.hook.AuthenticatedHook;
+import com.michelin.kafkactl.model.Resource;
+import com.michelin.kafkactl.model.request.ConnectorOffsets;
+import com.michelin.kafkactl.service.FormatService;
+import com.michelin.kafkactl.service.ResourceService;
+import io.micronaut.core.annotation.ReflectiveAccess;
+import jakarta.inject.Inject;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.List;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.Constructor;
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Option;
+import picocli.CommandLine.Parameters;
+
+/** Alter connector offsets subcommand. */
+@Command(
+        name = "alter-offsets",
+        headerHeading = "@|bold Usage|@:",
+        synopsisHeading = " ",
+        descriptionHeading = "%n@|bold Description|@: ",
+        description = "Alter or partially reset connector offsets.",
+        parameterListHeading = "%n@|bold Parameters|@:%n",
+        optionListHeading = "%n@|bold Options|@:%n",
+        commandListHeading = "%n@|bold Commands|@:%n",
+        usageHelpAutoWidth = true)
+public class ConnectorAlterOffsets extends AuthenticatedHook {
+    @Inject
+    @ReflectiveAccess
+    private ResourceService resourceService;
+
+    @Inject
+    @ReflectiveAccess
+    private FormatService formatService;
+
+    @Parameters(index = "0", description = "Connector name.")
+    public String connector;
+
+    @Option(
+            names = {"-f", "--file"},
+            description = "YAML file containing the connector offsets payload.",
+            required = true)
+    public File file;
+
+    @Override
+    public Integer onAuthSuccess() throws IOException {
+        ConnectorOffsets offsets = new Yaml(new Constructor(ConnectorOffsets.class, new LoaderOptions()))
+                .load(Files.readString(file.toPath()));
+
+        return resourceService
+                .alterConnectorOffsets(getNamespace(), connector, offsets, commandSpec)
+                .map(this::displayResponse)
+                .orElse(1);
+    }
+
+    private int displayResponse(Resource response) {
+        formatService.displayList(CONNECTOR_RESET_OFFSETS_RESPONSE, List.of(response), TABLE, commandSpec);
+        return 0;
+    }
+}

src/main/java/com/michelin/kafkactl/model/request/ConnectorOffsets.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package com.michelin.kafkactl.model.request;
+
+import io.micronaut.core.annotation.ReflectiveAccess;
+import java.util.List;
+import java.util.Map;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** Connector offsets alteration request. */
+@Data
+@ReflectiveAccess
+@NoArgsConstructor
+@AllArgsConstructor
+public class ConnectorOffsets {
+    private List<ConnectorOffset> offsets;
+
+    /** A connector partition and its new offset. */
+    @Data
+    @ReflectiveAccess
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class ConnectorOffset {
+        private Map<String, Object> partition;
+        private Map<String, Object> offset;
+    }
+}

src/main/java/com/michelin/kafkactl/service/ResourceService.java

@@ -33,6 +33,7 @@
 import com.michelin.kafkactl.model.Output;
 import com.michelin.kafkactl.model.Resource;
 import com.michelin.kafkactl.model.SubjectCompatibility;
+import com.michelin.kafkactl.model.request.ConnectorOffsets;
 import com.michelin.kafkactl.model.request.DeleteResourceRequest;
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference;
@@ -477,6 +478,32 @@ public Optional<Resource> resetConnectorOffsets(String namespace, String connect
         }
     }
 
+    /**
+     * Alter offsets for a given connector.
+     *
+     * @param namespace The namespace
+     * @param connector The connector name
+     * @param offsets The offsets payload
+     * @param commandSpec The command that triggered the action
+     * @return The resource
+     */
+    public Optional<Resource> alterConnectorOffsets(
+            String namespace, String connector, ConnectorOffsets offsets, CommandSpec commandSpec) {
+        try {
+            HttpResponse<Resource> response = namespacedClient.alterConnectorOffsets(
+                    namespace, connector, offsets, loginService.getAuthorization());
+
+            if (response.getStatus().equals(HttpStatus.NOT_FOUND)) {
+                throw new HttpClientResponseException(response.reason(), response);
+            }
+
+            return response.getBody();
+        } catch (HttpClientResponseException exception) {
+            formatService.displayError(exception, CONNECTOR, connector, commandSpec);
+            return Optional.empty();
+        }
+    }
+
     /**
      * Update the config of a given subject.
      *