Add support for multiple configuration contexts

README.md

@@ -26,6 +26,25 @@ It is recommended to install the bash/zsh completion script _kcctl_completion_:
 . kcctl_completion
 ```
 
+## Quickstart
+
+Before you can start using _kcctl_ you need to create a configuration context.
+A configuration context is a set of configuration parameters, grouped
+by a name.
+To create a configuration context named `local`, with the Kafka Connect cluster URL set to 
+`http://localhost:8083`, issue the following command
+
+```shell script
+kcctl set-context local --cluster http://localhost:8083
+```
+
+:exclamation: Note that certain commands will require additional parameters, like `bootstrap-servers` and 
+`offset-topic`.
+
+Type `kcctl info` to display some information about the Kafka connect cluster.
+The command will use the currently active context, `local` in this case, to
+resolve the cluster URL.
+
 ## Usage
 
 Display the help to learn about using _kcctl_:

kcctl_completion

@@ -401,14 +401,20 @@ function _picocli_kcctl_config_setcontext() {
 
   local commands=""
   local flag_opts=""
-  local arg_opts="--cluster"
+  local arg_opts="--cluster --bootstrap-servers --offset-topic"
 
   compopt +o default
 
   case ${prev_word} in
     --cluster)
       return
       ;;
+    --bootstrap-servers)
+      return
+      ;;
+    --offset-topic)
+      return
+      ;;
   esac
 
   if [[ "${curr_word}" == -* ]]; then

pom.xml

@@ -97,6 +97,17 @@
       <artifactId>quarkus-junit5</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>net.javacrumbs.json-unit</groupId>
+      <artifactId>json-unit-assertj</artifactId>
+      <version>2.27.0</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

src/main/java/dev/morling/kccli/command/ApplyCommand.java

@@ -48,7 +48,7 @@ public class ApplyCommand implements Callable<Integer> {
     @Override
     public Integer call() throws Exception {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         if (!file.exists()) {

src/main/java/dev/morling/kccli/command/ConnectorNamesCompletionCandidateCommand.java

@@ -34,7 +34,7 @@ public class ConnectorNamesCompletionCandidateCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         List<String> connectors = kafkaConnectApi.getConnectors();

src/main/java/dev/morling/kccli/command/DeleteConnectorCommand.java

@@ -37,7 +37,7 @@ public class DeleteConnectorCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         kafkaConnectApi.deleteConnector(name);

src/main/java/dev/morling/kccli/command/DescribeConnectorCommand.java

@@ -63,7 +63,7 @@ public class DescribeConnectorCommand implements Callable<Integer> {
     @Override
     public Integer call() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         if (includeTasksConfig) {

src/main/java/dev/morling/kccli/command/GetConnectorsCommand.java

@@ -44,7 +44,7 @@ public class GetConnectorsCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         List<String> connectors = kafkaConnectApi.getConnectors();

src/main/java/dev/morling/kccli/command/GetContextCommand.java

@@ -28,7 +28,7 @@ public class GetContextCommand implements Runnable {
 
     @Override
     public void run() {
-        String clusterUri = context.getCluster().toASCIIString();
+        String clusterUri = context.getContext().getCluster().toASCIIString();
         System.out.println("Current context is set to " + clusterUri);
     }
 }

src/main/java/dev/morling/kccli/command/GetLoggerCommand.java

@@ -54,7 +54,7 @@ public class GetLoggerCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         String[][] data;

src/main/java/dev/morling/kccli/command/GetLoggersCommand.java

@@ -47,7 +47,7 @@ public class GetLoggersCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         ObjectNode connectorLoggers = kafkaConnectApi.getLoggers("");

src/main/java/dev/morling/kccli/command/GetPluginsCommand.java

@@ -41,7 +41,7 @@ public class GetPluginsCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         List<ConnectorPlugin> connectorPlugins = kafkaConnectApi.getConnectorPlugins();

src/main/java/dev/morling/kccli/command/InfoCommand.java

@@ -33,11 +33,11 @@ public class InfoCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         KafkaConnectInfo workerInfo = kafkaConnectApi.getWorkerInfo();
-        System.out.println("URL:               " + context.getCluster());
+        System.out.println("URL:               " + context.getContext().getCluster());
         System.out.println("Version:           " + workerInfo.version);
         System.out.println("Commit:            " + workerInfo.commit);
         System.out.println("Kafka Cluster ID:  " + workerInfo.kafka_cluster_id);

src/main/java/dev/morling/kccli/command/LoggerNamesCompletionCandidateCommand.java

@@ -38,7 +38,7 @@ public class LoggerNamesCompletionCandidateCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         ObjectNode connectorLoggers = kafkaConnectApi.getLoggers("");

src/main/java/dev/morling/kccli/command/PatchConnectorCommand.java

@@ -61,7 +61,7 @@ public class PatchConnectorCommand implements Callable<Integer> {
     public Integer call() throws JsonProcessingException {
 
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         Map<String, String> connectorParameters = kafkaConnectApi.getConnectorConfig(name);

src/main/java/dev/morling/kccli/command/PatchLogLevelCommand.java

@@ -45,7 +45,7 @@ public class PatchLogLevelCommand implements Callable {
     @Override
     public Object call() throws Exception {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         ObjectMapper mapper = new ObjectMapper();

src/main/java/dev/morling/kccli/command/PauseConnectorCommand.java

@@ -37,7 +37,7 @@ public class PauseConnectorCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         kafkaConnectApi.pauseConnector(name);

src/main/java/dev/morling/kccli/command/RestartConnectorCommand.java

@@ -37,7 +37,7 @@ public class RestartConnectorCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         kafkaConnectApi.restartConnector(name);

src/main/java/dev/morling/kccli/command/RestartTaskCommand.java

@@ -37,7 +37,7 @@ public class RestartTaskCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         String[] parts = name.split("\\/");

src/main/java/dev/morling/kccli/command/ResumeConnectorCommand.java

@@ -37,7 +37,7 @@ public class ResumeConnectorCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         kafkaConnectApi.resumeConnector(name);

src/main/java/dev/morling/kccli/command/SetContextCommand.java

@@ -15,20 +15,32 @@
  */
 package dev.morling.kccli.command;
 
+import java.net.URI;
+
+import dev.morling.kccli.service.Context;
 import dev.morling.kccli.util.ConfigurationContext;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Option;
+import picocli.CommandLine.Parameters;
 
 @Command(name = "set-context", description = "Set the context to cluster provided in arguments")
 public class SetContextCommand implements Runnable {
+    @Parameters(index = "0", description = "Context name")
+    String contextName;
 
-    @Option(names = { "--cluster" }, description = "URL of the Kafka Connect cluster to connect to", required = true)
+    @Option(names = { "--cluster" }, defaultValue = "http://localhost:8083", description = "URL of the Kafka Connect cluster to connect to")
     String cluster;
 
+    @Option(names = { "--bootstrap-servers" }, defaultValue = "localhost:9092", description = "Comma-separated list of Kafka broker URLs")
+    String bootstrapServers;
+
+    @Option(names = { "--offset-topic" }, description = "Name of the offset topic")
+    String offsetTopic;
+
     @Override
     public void run() {
         ConfigurationContext context = new ConfigurationContext();
-        context.setConfiguration(cluster);
-        System.out.println("Successfully set context to " + cluster);
+        context.setContext(contextName, new Context(URI.create(cluster), bootstrapServers, offsetTopic));
+        System.out.println("Using context " + contextName);
     }
 }

src/main/java/dev/morling/kccli/command/TaskNamesCompletionCandidateCommand.java

@@ -37,7 +37,7 @@ public class TaskNamesCompletionCandidateCommand implements Runnable {
     @Override
     public void run() {
         KafkaConnectApi kafkaConnectApi = RestClientBuilder.newBuilder()
-                .baseUri(context.getCluster())
+                .baseUri(context.getContext().getCluster())
                 .build(KafkaConnectApi.class);
 
         List<String> connectors = kafkaConnectApi.getConnectors();

src/main/java/dev/morling/kccli/service/Configuration.java

@@ -0,0 +1,48 @@
+/*
+ *  Copyright 2021 The original authors
+ *
+ *  Licensed 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 dev.morling.kccli.service;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonAnyGetter;
+import com.fasterxml.jackson.annotation.JsonAnySetter;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class Configuration {
+    private final String currentContext;
+    private final Map<String, Context> configurationContexts = new LinkedHashMap<>();
+
+    public Configuration(@JsonProperty("currentContext") String currentContext) {
+        this.currentContext = currentContext;
+    }
+
+    public String getCurrentContext() {
+        return currentContext;
+    }
+
+    @JsonAnyGetter
+    public Map<String, Context> configurationContexts() {
+        return configurationContexts;
+    }
+
+    @JsonAnySetter
+    public Configuration addConfigurationContext(String contextName, Context configuration) {
+        configurationContexts.put(contextName, configuration);
+
+        return this;
+    }
+}

src/main/java/dev/morling/kccli/service/Context.java

@@ -0,0 +1,51 @@
+/*
+ *  Copyright 2021 The original authors
+ *
+ *  Licensed 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 dev.morling.kccli.service;
+
+import java.net.URI;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class Context {
+    private final URI cluster;
+    private final String bootstrapServers;
+    private final String offsetTopic;
+
+    public Context(
+                   @JsonProperty("cluster") URI cluster,
+                   @JsonProperty("bootstrapServers") String bootstrapServers,
+                   @JsonProperty("offsetTopic") String offsetTopic) {
+        this.cluster = cluster;
+        this.bootstrapServers = bootstrapServers;
+        this.offsetTopic = offsetTopic;
+    }
+
+    public URI getCluster() {
+        return cluster;
+    }
+
+    public String getBootstrapServers() {
+        return bootstrapServers;
+    }
+
+    public String getOffsetTopic() {
+        return offsetTopic;
+    }
+
+    public static Context defaultContext() {
+        return new Context(URI.create("http://localhost:8083"), null, null);
+    }
+}