implement more end points

Author: Crim

src/main/java/org/sourcelab/kafka/connect/apiclient/ApiClient.java

@@ -6,12 +6,15 @@
 import org.sourcelab.kafka.connect.apiclient.request.Request;
 import org.sourcelab.kafka.connect.apiclient.request.RequestErrorResponse;
 import org.sourcelab.kafka.connect.apiclient.request.delete.DeleteConnector;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorPlugin;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorPluginConfigDefinition;
 import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorStatus;
 import org.sourcelab.kafka.connect.apiclient.request.dto.NewConnectorDefinition;
 import org.sourcelab.kafka.connect.apiclient.request.dto.Task;
 import org.sourcelab.kafka.connect.apiclient.request.dto.TaskStatus;
 import org.sourcelab.kafka.connect.apiclient.request.get.GetConnector;
 import org.sourcelab.kafka.connect.apiclient.request.get.GetConnectorConfig;
+import org.sourcelab.kafka.connect.apiclient.request.get.GetConnectorPlugins;
 import org.sourcelab.kafka.connect.apiclient.request.get.GetConnectorStatus;
 import org.sourcelab.kafka.connect.apiclient.request.get.GetConnectorTaskStatus;
 import org.sourcelab.kafka.connect.apiclient.request.get.GetConnectorTasks;
@@ -22,6 +25,7 @@
 import org.sourcelab.kafka.connect.apiclient.request.post.PostConnectorTaskRestart;
 import org.sourcelab.kafka.connect.apiclient.request.put.PutConnectorConfig;
 import org.sourcelab.kafka.connect.apiclient.request.put.PutConnectorPause;
+import org.sourcelab.kafka.connect.apiclient.request.put.PutConnectorPluginConfigValidate;
 import org.sourcelab.kafka.connect.apiclient.request.put.PutConnectorResume;
 import org.sourcelab.kafka.connect.apiclient.rest.HttpClientRestClient;
 import org.sourcelab.kafka.connect.apiclient.rest.InvalidRequestException;
@@ -231,6 +235,30 @@ public Boolean restartConnectorTask(final String connectorName, final int taskId
         return submitRequest(new PostConnectorTaskRestart(connectorName, taskId));
     }
 
+    /**
+     * Return a list of connector plugins installed in the Kafka Connect cluster.
+     * https://docs.confluent.io/current/connect/restapi.html#get--connector-plugins-
+     *
+     * @return Collection of available connector plugins.
+     */
+    public Collection<ConnectorPlugin> getConnectorPlugins() {
+        return submitRequest(new GetConnectorPlugins());
+    }
+
+    /**
+     * Validate the provided configuration values against the configuration definition. This API performs per config
+     * validation, returns suggested values and error messages during validation.
+     * https://docs.confluent.io/current/connect/restapi.html#put--connector-plugins-(string-name)-config-validate
+     *
+     * @param configDefinition Defines the configuration to validate.
+     * @return Results of the validation.
+     */
+    public String validateConnectorPluginConfig(final ConnectorPluginConfigDefinition configDefinition) {
+        return submitRequest(
+            new PutConnectorPluginConfigValidate(configDefinition.getName(), configDefinition.getConfig())
+        );
+    }
+
     /**
      * package protected for access in tests.
      * @return Rest Client.

src/main/java/org/sourcelab/kafka/connect/apiclient/request/dto/ConnectorPlugin.java

@@ -2,14 +2,27 @@
 
 import com.fasterxml.jackson.annotation.JsonAlias;
 
+/**
+ * Represents details about a Connector Plugin.
+ */
 public class ConnectorPlugin {
     @JsonAlias("class")
     private String className;
+    private String type;
+    private String version;
 
-    private String getClassName() {
+    public String getClassName() {
         return className;
     }
 
+    public String getType() {
+        return type;
+    }
+
+    public String getVersion() {
+        return version;
+    }
+
     @Override
     public String toString() {
         return "ConnectorPlugin{"

src/main/java/org/sourcelab/kafka/connect/apiclient/request/dto/ConnectorPluginConfigDefinition.java

@@ -0,0 +1,78 @@
+package org.sourcelab.kafka.connect.apiclient.request.dto;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Defines a Connector Plugin Configuration.
+ */
+public class ConnectorPluginConfigDefinition {
+    private final String name;
+    private final Map<String, String> config;
+
+    /**
+     * Constructor
+     * @param connectorPluginName Name of Connector Plugin.
+     * @param config Configuration values for connector.
+     */
+    public ConnectorPluginConfigDefinition(final String connectorPluginName, final Map<String, String> config) {
+        this.name = connectorPluginName;
+        this.config = Collections.unmodifiableMap(new HashMap<>(config));
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public Map<String, String> getConfig() {
+        return config;
+    }
+
+    public static ConnectorPluginConfigDefinition.Builder newBuilder() {
+        return new ConnectorPluginConfigDefinition.Builder();
+    }
+
+    @Override
+    public String toString() {
+        return "ConnectorPluginConfigDefinition{"
+            + "name='" + name + '\''
+            + ", config=" + config
+            + '}';
+    }
+
+    /**
+     * Builder for ConnectorPluginConfigDefinition.
+     */
+    public static final class Builder {
+        private String name;
+        private Map<String, String> config = new HashMap<>();
+
+        private Builder() {
+        }
+
+        public ConnectorPluginConfigDefinition.Builder withName(final String name) {
+            this.name = name;
+            return this;
+        }
+
+        public ConnectorPluginConfigDefinition.Builder withConfig(final Map<String, String> config) {
+            this.config = new HashMap<>(config);
+            return this;
+        }
+
+        public ConnectorPluginConfigDefinition.Builder withConfig(final String key, final String value) {
+            this.config.put(key, value);
+            return this;
+        }
+
+        public ConnectorPluginConfigDefinition.Builder withConfig(final String key, final Object value) {
+            this.config.put(key, value.toString());
+            return this;
+        }
+
+        public ConnectorPluginConfigDefinition build() {
+            return new ConnectorPluginConfigDefinition(name, config);
+        }
+    }
+}

src/main/java/org/sourcelab/kafka/connect/apiclient/request/dto/ConnectorPluginConfigValidation.java

@@ -0,0 +1,8 @@
+package org.sourcelab.kafka.connect.apiclient.request.dto;
+
+/**
+ * Represents the results of validating a Connector Plugin configuration.
+ */
+public class ConnectorPluginConfigValidation {
+
+}

src/main/java/org/sourcelab/kafka/connect/apiclient/request/put/PutConnectorConfig.java

@@ -16,7 +16,7 @@ public class PutConnectorConfig implements PutRequest<ConnectorDefinition> {
 
     public PutConnectorConfig(final String name, final Map<String, String> config) {
         this.name = name;
-        this.config = Collections.unmodifiableMap(new HashMap(config));
+        this.config = Collections.unmodifiableMap(new HashMap<>(config));
     }
 
     @Override

src/main/java/org/sourcelab/kafka/connect/apiclient/request/put/PutConnectorPluginConfigValidate.java

@@ -0,0 +1,42 @@
+package org.sourcelab.kafka.connect.apiclient.request.put;
+
+import com.sun.xml.internal.rngom.util.Uri;
+import org.sourcelab.kafka.connect.apiclient.request.JacksonFactory;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorDefinition;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class PutConnectorPluginConfigValidate implements PutRequest<String> {
+    private final String connectorPluginName;
+    private final Map<String, String> config;
+
+    public PutConnectorPluginConfigValidate(final String connectorPluginName, final Map<String, String> config) {
+        if (connectorPluginName == null) {
+            throw new NullPointerException("ConnectorPluginName parameter may not be null!");
+        }
+        if (config == null) {
+            throw new NullPointerException("config parameter may not be null!");
+        }
+        this.connectorPluginName = connectorPluginName;
+        this.config = Collections.unmodifiableMap(new HashMap<>(config));
+    }
+
+    @Override
+    public String getApiEndpoint() {
+        return "/connector-plugins/" + Uri.escapeDisallowedChars(connectorPluginName) + "/config/validate";
+    }
+
+    @Override
+    public Object getRequestBody() {
+        return config;
+    }
+
+    @Override
+    public String parseResponse(final String responseStr) throws IOException {
+        //return JacksonFactory.newInstance().readValue(responseStr, ConnectorDefinition.class);
+        return responseStr;
+    }
+}

src/main/java/org/sourcelab/kafka/connect/apiclient/rest/HttpClientRestClient.java

@@ -273,12 +273,8 @@ private <T> T submitPutRequest(final String url, final Object requestBody, final
             // Conditionally add content-type header?
             put.addHeader(new BasicHeader("Content-Type", "application/json"));
 
-            // Define required auth params
-            final List<NameValuePair> params = new ArrayList<>();
-
-            // Convert to Json
+            // Convert to Json and submit as payload.
             final String jsonPayloadStr = JacksonFactory.newInstance().writeValueAsString(requestBody);
-
             put.setEntity(new StringEntity(jsonPayloadStr));
 
             logger.info("Executing request {} with {}", put.getRequestLine(), jsonPayloadStr);

src/test/java/org/sourcelab/kafka/connect/apiclient/ApiClientTest.java

@@ -5,6 +5,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorDefinition;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorPluginConfigDefinition;
 import org.sourcelab.kafka.connect.apiclient.request.dto.NewConnectorDefinition;
 
 import java.util.HashMap;
@@ -147,10 +148,32 @@ public void testGetConnectorTaskStatus() {
     }
 
     /**
-     * Test retrieving status about a specific task for a running connector.
+     * Test restarting a specific task for a running connector.
      */
     @Test
     public void testRestartConnectorTask() {
         logger.info("Result: {}", apiClient.restartConnectorTask(connectorName, 0));
     }
+
+    /**
+     * Test retrieving available connector plugins.
+     */
+    @Test
+    public void testGetConnectorPlugins() {
+        logger.info("Result: {}", apiClient.getConnectorPlugins());
+    }
+
+    /**
+     * Test retrieving available connector plugins.
+     */
+    @Test
+    public void testValidateConnectorPluginConfig() {
+        logger.info("Result: {}", apiClient.validateConnectorPluginConfig(ConnectorPluginConfigDefinition.newBuilder()
+            .withName("VerifiableSourceConnector")
+            .withConfig("connector.class", "org.apache.kafka.connect.tools.VerifiableSourceConnector")
+            .withConfig("tasks.max", 3)
+            .withConfig("topics", "test-topic")
+            .build()
+        ));
+    }
 }

src/test/resources/mockResponses/getConnectorPlugins.json

@@ -0,0 +1,9 @@
+[{
+  "class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
+  "type": "sink",
+  "version": "1.0.0-cp1"
+}, {
+  "class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
+  "type": "source",
+  "version": "1.0.0-cp1"
+}]