diff --git a/README.md b/README.md index c54c19f1..a3d84e6b 100644 --- a/README.md +++ b/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=] -f= [-n=] + + +Description: Alter or partially reset connector offsets. + +Parameters: + Connector name. + +Options: + -c, --context= + Override context defined in config. + -f, --file= YAML file containing the connector offsets payload. + -h, --help Show this help message and exit. + -n, --namespace= + 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. diff --git a/src/main/java/com/michelin/kafkactl/client/NamespacedResourceClient.java b/src/main/java/com/michelin/kafkactl/client/NamespacedResourceClient.java index dac35616..c3ed3fef 100644 --- a/src/main/java/com/michelin/kafkactl/client/NamespacedResourceClient.java +++ b/src/main/java/com/michelin/kafkactl/client/NamespacedResourceClient.java @@ -19,6 +19,7 @@ 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; @@ -26,6 +27,7 @@ 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 updateSubjectConfig( HttpResponse 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 alterConnectorOffsets( + String namespace, String connector, @Body ConnectorOffsets offsets, @Header("Authorization") String token); + /** * List all available connect clusters for vaulting. * diff --git a/src/main/java/com/michelin/kafkactl/command/connector/Connector.java b/src/main/java/com/michelin/kafkactl/command/connector/Connector.java index 106b37fc..14337968 100644 --- a/src/main/java/com/michelin/kafkactl/command/connector/Connector.java +++ b/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|@: ", diff --git a/src/main/java/com/michelin/kafkactl/command/connector/ConnectorAlterOffsets.java b/src/main/java/com/michelin/kafkactl/command/connector/ConnectorAlterOffsets.java new file mode 100644 index 00000000..bdf56d76 --- /dev/null +++ b/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; + } +} diff --git a/src/main/java/com/michelin/kafkactl/model/request/ConnectorOffsets.java b/src/main/java/com/michelin/kafkactl/model/request/ConnectorOffsets.java new file mode 100644 index 00000000..32b5ed42 --- /dev/null +++ b/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 offsets; + + /** A connector partition and its new offset. */ + @Data + @ReflectiveAccess + @NoArgsConstructor + @AllArgsConstructor + public static class ConnectorOffset { + private Map partition; + private Map offset; + } +} diff --git a/src/main/java/com/michelin/kafkactl/service/ResourceService.java b/src/main/java/com/michelin/kafkactl/service/ResourceService.java index 9708f784..db90dd9f 100644 --- a/src/main/java/com/michelin/kafkactl/service/ResourceService.java +++ b/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 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 alterConnectorOffsets( + String namespace, String connector, ConnectorOffsets offsets, CommandSpec commandSpec) { + try { + HttpResponse 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. * diff --git a/src/test/java/com/michelin/kafkactl/command/connector/ConnectorAlterOffsetsTest.java b/src/test/java/com/michelin/kafkactl/command/connector/ConnectorAlterOffsetsTest.java new file mode 100644 index 00000000..ec45fbd7 --- /dev/null +++ b/src/test/java/com/michelin/kafkactl/command/connector/ConnectorAlterOffsetsTest.java @@ -0,0 +1,129 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.Mockito.argThat; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.michelin.kafkactl.model.Resource; +import com.michelin.kafkactl.property.KafkactlProperties; +import com.michelin.kafkactl.service.ConfigService; +import com.michelin.kafkactl.service.FormatService; +import com.michelin.kafkactl.service.LoginService; +import com.michelin.kafkactl.service.ResourceService; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import picocli.CommandLine; + +@ExtendWith(MockitoExtension.class) +class ConnectorAlterOffsetsTest { + @Mock + LoginService loginService; + + @Mock + KafkactlProperties kafkactlProperties; + + @Mock + ResourceService resourceService; + + @Mock + FormatService formatService; + + @Mock + ConfigService configService; + + @InjectMocks + ConnectorAlterOffsets connectorAlterOffsets; + + @TempDir + Path tempDir; + + @Test + void shouldAlterConnectorOffsets() throws IOException { + Path offsetsFile = writeOffsetsFile(); + Resource response = Resource.builder() + .kind(CONNECTOR_RESET_OFFSETS_RESPONSE) + .metadata(Resource.Metadata.builder().name("my-connector").build()) + .status(Map.of("code", "RESET")) + .build(); + + when(configService.isCurrentContextValid()).thenReturn(true); + when(loginService.doAuthenticate(any(), anyBoolean())).thenReturn(true); + when(resourceService.alterConnectorOffsets(any(), any(), any(), any())).thenReturn(Optional.of(response)); + + CommandLine cmd = new CommandLine(connectorAlterOffsets); + int code = cmd.execute("my-connector", "-f", offsetsFile.toString(), "-n", "namespace"); + + assertEquals(0, code); + verify(resourceService) + .alterConnectorOffsets( + eq("namespace"), + eq("my-connector"), + argThat(request -> request.getOffsets().size() == 1 + && request.getOffsets() + .getFirst() + .getPartition() + .equals(Map.of("kafka_topic", "topic1", "kafka_partition", 0)) + && request.getOffsets().getFirst().getOffset() == null), + eq(cmd.getCommandSpec())); + verify(formatService) + .displayList(CONNECTOR_RESET_OFFSETS_RESPONSE, List.of(response), TABLE, cmd.getCommandSpec()); + } + + @Test + void shouldReturnFailureWhenResponseIsEmpty() throws IOException { + Path offsetsFile = writeOffsetsFile(); + + when(configService.isCurrentContextValid()).thenReturn(true); + when(loginService.doAuthenticate(any(), anyBoolean())).thenReturn(true); + when(resourceService.alterConnectorOffsets(any(), any(), any(), any())).thenReturn(Optional.empty()); + + CommandLine cmd = new CommandLine(connectorAlterOffsets); + int code = cmd.execute("my-connector", "-f", offsetsFile.toString(), "-n", "namespace"); + + assertEquals(1, code); + } + + private Path writeOffsetsFile() throws IOException { + return Files.writeString(tempDir.resolve("offsets.yaml"), """ + offsets: + - partition: + kafka_topic: topic1 + kafka_partition: 0 + offset: null + """); + } +} diff --git a/src/test/java/com/michelin/kafkactl/command/connector/ConnectorTest.java b/src/test/java/com/michelin/kafkactl/command/connector/ConnectorTest.java index 55e3d826..03660c53 100644 --- a/src/test/java/com/michelin/kafkactl/command/connector/ConnectorTest.java +++ b/src/test/java/com/michelin/kafkactl/command/connector/ConnectorTest.java @@ -114,6 +114,18 @@ void shouldDisplayResetOffsetsHelpWithoutParentParameters() { assertTrue(sw.toString().contains("Reset connector offsets.")); } + @Test + void shouldDisplayAlterOffsetsHelpWithoutParentParameters() { + CommandLine cmd = new CommandLine(connector); + StringWriter sw = new StringWriter(); + cmd.setOut(new PrintWriter(sw)); + + int code = cmd.execute("alter-offsets", "--help"); + + assertEquals(0, code); + assertTrue(sw.toString().contains("Alter or partially reset connector offsets.")); + } + @Test void shouldNotChangeStateWhenEmptyConnectorsList() { when(configService.isCurrentContextValid()).thenReturn(true); diff --git a/src/test/java/com/michelin/kafkactl/service/ResourceServiceTest.java b/src/test/java/com/michelin/kafkactl/service/ResourceServiceTest.java index dee784ab..a64ef53c 100644 --- a/src/test/java/com/michelin/kafkactl/service/ResourceServiceTest.java +++ b/src/test/java/com/michelin/kafkactl/service/ResourceServiceTest.java @@ -48,6 +48,7 @@ import com.michelin.kafkactl.model.ApiResource; 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.micronaut.http.HttpResponse; import io.micronaut.http.HttpStatus; @@ -1686,6 +1687,61 @@ void shouldResetConnectorOffsetsFail() { verify(formatService).displayError(exception, CONNECTOR, "connector", cmd.getCommandSpec()); } + @Test + void shouldAlterConnectorOffsets() { + Resource response = Resource.builder() + .kind("ConnectorResetOffsetsResponse") + .metadata(Resource.Metadata.builder().name("connector").build()) + .status(Map.of("code", "RESET")) + .build(); + ConnectorOffsets offsets = new ConnectorOffsets( + List.of(new ConnectorOffsets.ConnectorOffset(Map.of("partition", 0), Map.of("offset", 10)))); + CommandLine cmd = new CommandLine(new Kafkactl()); + + when(namespacedClient.alterConnectorOffsets(any(), any(), any(), any())).thenReturn(HttpResponse.ok(response)); + + Optional actual = + resourceService.alterConnectorOffsets("namespace", "connector", offsets, cmd.getCommandSpec()); + + assertEquals(Optional.of(response), actual); + verify(namespacedClient).alterConnectorOffsets("namespace", "connector", offsets, null); + } + + @Test + void shouldAlterConnectorOffsetsNotFound() { + ConnectorOffsets offsets = new ConnectorOffsets(List.of()); + CommandLine cmd = new CommandLine(new Kafkactl()); + + when(namespacedClient.alterConnectorOffsets(any(), any(), any(), any())).thenReturn(HttpResponse.notFound()); + + Optional actual = + resourceService.alterConnectorOffsets("namespace", "connector", offsets, cmd.getCommandSpec()); + + assertTrue(actual.isEmpty()); + verify(formatService) + .displayError( + argThat(exception -> exception.getStatus().equals(HttpStatus.NOT_FOUND) + && exception.getMessage().equals("Not Found")), + eq(CONNECTOR), + eq("connector"), + eq(cmd.getCommandSpec())); + } + + @Test + void shouldAlterConnectorOffsetsFail() { + ConnectorOffsets offsets = new ConnectorOffsets(List.of()); + CommandLine cmd = new CommandLine(new Kafkactl()); + HttpClientResponseException exception = new HttpClientResponseException("error", HttpResponse.serverError()); + + when(namespacedClient.alterConnectorOffsets(any(), any(), any(), any())).thenThrow(exception); + + Optional actual = + resourceService.alterConnectorOffsets("namespace", "connector", offsets, cmd.getCommandSpec()); + + assertTrue(actual.isEmpty()); + verify(formatService).displayError(exception, CONNECTOR, "connector", cmd.getCommandSpec()); + } + @Test void shouldUpdateSubjectConfig() { Resource updateSubjectConfigResource = Resource.builder()