Merge pull request #25 from SourceLabOrg/sp/issue24

[ISSUE-24] Initial support for KIP-465

Author: Crim

CHANGELOG.md

@@ -2,6 +2,57 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## 2.1.0 (06/26/2019)
+
+### New Features
+- Added support to retrieve information about the Kafka-Connect server being queried.
+
+```java
+    /**
+     * Retrieve details about the Kafka-Connect service itself.
+     * @return ConnectServerVersion
+     */
+    public ConnectServerVersion getConnectServerVersion()
+```
+
+- Added support for [Expanded Connectors API Endpoint KIP-465](https://cwiki.apache.org/confluence/display/KAFKA/KIP-465%3A+Add+Consolidated+Connector+Endpoint+to+Connect+REST+API).
+
+Only supported by Kafka-Connect servers running version 2.3.0+ the following methods were added:
+
+```java
+    /**
+     * Get a list of deployed connectors, including the status for each connector.
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors
+     *
+     * Requires Kafka-Connect 2.3.0+
+     *
+     * @return All deployed connectors, and their respective statuses.
+     */
+    public ConnectorsWithExpandedStatus getConnectorsWithExpandedStatus()
+
+    /**
+     * Get a list of deployed connectors, including the definition for each connector.
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors
+     *
+     * Requires Kafka-Connect 2.3.0+
+     *
+     * @return All deployed connectors, and their respective definition.
+     */
+    public ConnectorsWithExpandedInfo getConnectorsWithExpandedInfo()
+
+    /**
+     * Get a list of deployed connectors, including all metadata available.
+     * Currently includes both 'info' {@see getConnectorsWithExpandedInfo} and 'status' {@see getConnectorsWithExpandedStatus}
+     * metadata.
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors
+     *
+     * Requires Kafka-Connect 2.3.0+
+     *
+     * @return All deployed connectors, and their respective metadata.
+     */
+    public ConnectorsWithExpandedMetadata getConnectorsWithAllExpandedMetadata()
+```
+
 ## 2.0.2 (06/06/2019)
 
 ### Internal Dependency Updates

README.md

@@ -12,7 +12,7 @@ This client library is released on Maven Central.  Add a new dependency to your
 <dependency>
     <groupId>org.sourcelab</groupId>
     <artifactId>kafka-connect-client</artifactId>
-    <version>2.0.2</version>
+    <version>2.1.0</version>
 </dependency>
 ```
 

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>org.sourcelab</groupId>
     <artifactId>kafka-connect-client</artifactId>
-    <version>2.0.2</version>
+    <version>2.1.0</version>
     <packaging>jar</packaging>
 
     <!-- Require Maven 3.3.9 -->

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

@@ -17,28 +17,38 @@
 
 package org.sourcelab.kafka.connect.apiclient;
 
+import com.fasterxml.jackson.databind.exc.MismatchedInputException;
 import org.apache.http.HttpStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.sourcelab.kafka.connect.apiclient.exception.ResponseParseException;
 import org.sourcelab.kafka.connect.apiclient.request.JacksonFactory;
 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.ConnectServerVersion;
 import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorDefinition;
 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.ConnectorPluginConfigValidationResults;
 import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorStatus;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorsWithExpandedInfo;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorsWithExpandedMetadata;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorsWithExpandedStatus;
 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.GetConnectServerVersion;
 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;
 import org.sourcelab.kafka.connect.apiclient.request.get.GetConnectors;
+import org.sourcelab.kafka.connect.apiclient.request.get.GetConnectorsExpandAllDetails;
+import org.sourcelab.kafka.connect.apiclient.request.get.GetConnectorsExpandInfo;
+import org.sourcelab.kafka.connect.apiclient.request.get.GetConnectorsExpandStatus;
 import org.sourcelab.kafka.connect.apiclient.request.post.PostConnector;
 import org.sourcelab.kafka.connect.apiclient.request.post.PostConnectorRestart;
 import org.sourcelab.kafka.connect.apiclient.request.post.PostConnectorTaskRestart;
@@ -59,7 +69,7 @@
 /**
  * API Client for interacting with the Kafka-Connect Rest Endpoint.
  * Official Rest Endpoint documentation can be found here:
- *   https://docs.confluent.io/current/connect/restapi.html
+ *   https://docs.confluent.io/current/connect/references/restapi.html
  */
 public class KafkaConnectClient {
     private static final Logger logger = LoggerFactory.getLogger(KafkaConnectClient.class);
@@ -92,7 +102,7 @@ public KafkaConnectClient(final Configuration configuration) {
     /**
      * Constructor for injecting a RestClient implementation.
      * Typically only used in testing.
-     * @param configuration Pardot Api Configuration.
+     * @param configuration Api Client Configuration.
      * @param restClient RestClient implementation to use.
      */
     public KafkaConnectClient(final Configuration configuration, final RestClient restClient) {
@@ -101,18 +111,64 @@ public KafkaConnectClient(final Configuration configuration, final RestClient re
     }
 
     /**
-     * Get a list of active connectors.
-     * https://docs.confluent.io/current/connect/restapi.html#get--connectors
+     * Retrieve details about the Kafka-Connect service itself.
+     * @return ConnectServerVersion
+     */
+    public ConnectServerVersion getConnectServerVersion() {
+        return submitRequest(new GetConnectServerVersion());
+    }
+
+    /**
+     * Get a list of deployed connectors.
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors
      *
      * @return Collection of connector names currently deployed.
      */
     public Collection<String> getConnectors() {
         return submitRequest(new GetConnectors());
     }
 
+    /**
+     * Get a list of deployed connectors, including the status for each connector.
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors
+     *
+     * Requires Kafka-Connect 2.3.0+
+     *
+     * @return All deployed connectors, and their respective statuses.
+     */
+    public ConnectorsWithExpandedStatus getConnectorsWithExpandedStatus() {
+        return submitRequest(new GetConnectorsExpandStatus());
+    }
+
+    /**
+     * Get a list of deployed connectors, including the definition for each connector.
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors
+     *
+     * Requires Kafka-Connect 2.3.0+
+     *
+     * @return All deployed connectors, and their respective definition.
+     */
+    public ConnectorsWithExpandedInfo getConnectorsWithExpandedInfo() {
+        return submitRequest(new GetConnectorsExpandInfo());
+    }
+
+    /**
+     * Get a list of deployed connectors, including all metadata available.
+     * Currently includes both 'info' {@see getConnectorsWithExpandedInfo} and 'status' {@see getConnectorsWithExpandedStatus}
+     * metadata.
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors
+     *
+     * Requires Kafka-Connect 2.3.0+
+     *
+     * @return All deployed connectors, and their respective metadata.
+     */
+    public ConnectorsWithExpandedMetadata getConnectorsWithAllExpandedMetadata() {
+        return submitRequest(new GetConnectorsExpandAllDetails());
+    }
+
     /**
      * Get information about the connector.
-     * https://docs.confluent.io/current/connect/restapi.html#get--connectors-(string-name)
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors-(string-name)
      * @param connectorName Name of connector.
      * @return Connector details.
      */
@@ -122,7 +178,7 @@ public ConnectorDefinition getConnector(final String connectorName) {
 
     /**
      * Get the configuration for the connector.
-     * https://docs.confluent.io/current/connect/restapi.html#get--connectors-(string-name)-config
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors-(string-name)-config
      * @param connectorName Name of connector.
      * @return Configuration for connector.
      */
@@ -132,7 +188,7 @@ public Map<String, String> getConnectorConfig(final String connectorName) {
 
     /**
      * Get the status of specified connector by name.
-     * https://docs.confluent.io/current/connect/restapi.html#get--connectors-(string-name)-config
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors-(string-name)-config
      *
      * @param connectorName Name of connector.
      * @return Status details of the connector.
@@ -143,7 +199,7 @@ public ConnectorStatus getConnectorStatus(final String connectorName) {
 
     /**
      * Create a new connector, returning the current connector info if successful.
-     * https://docs.confluent.io/current/connect/restapi.html#post--connectors
+     * https://docs.confluent.io/current/connect/references/restapi.html#post--connectors
      *
      * @param connectorDefinition Defines the new connector to deploy
      * @return connector info.
@@ -154,7 +210,7 @@ public ConnectorDefinition addConnector(final NewConnectorDefinition connectorDe
 
     /**
      * Update a connector's configuration.
-     * https://docs.confluent.io/current/connect/restapi.html#put--connectors-(string-name)-config
+     * https://docs.confluent.io/current/connect/references/restapi.html#put--connectors-(string-name)-config
      *
      * @param connectorName Name of connector to update.
      * @param config Configuration values to set.
@@ -166,7 +222,7 @@ public ConnectorDefinition updateConnectorConfig(final String connectorName, fin
 
     /**
      * Restart a connector.
-     * https://docs.confluent.io/current/connect/restapi.html#post--connectors-(string-name)-restart
+     * https://docs.confluent.io/current/connect/references/restapi.html#post--connectors-(string-name)-restart
      *
      * @param connectorName Name of connector to restart.
      * @return Boolean true if success.
@@ -177,7 +233,7 @@ public Boolean restartConnector(final String connectorName) {
 
     /**
      * Pause a connector.
-     * https://docs.confluent.io/current/connect/restapi.html#put--connectors-(string-name)-pause
+     * https://docs.confluent.io/current/connect/references/restapi.html#put--connectors-(string-name)-pause
      *
      * @param connectorName Name of connector to pause.
      * @return Boolean true if success.
@@ -188,7 +244,7 @@ public Boolean pauseConnector(final String connectorName) {
 
     /**
      * Resume a connector.
-     * https://docs.confluent.io/current/connect/restapi.html#put--connectors-(string-name)-resume
+     * https://docs.confluent.io/current/connect/references/restapi.html#put--connectors-(string-name)-resume
      *
      * @param connectorName Name of connector to resume.
      * @return Boolean true if success.
@@ -199,7 +255,7 @@ public Boolean resumeConnector(final String connectorName) {
 
     /**
      * Resume a connector.
-     * https://docs.confluent.io/current/connect/restapi.html#put--connectors-(string-name)-resume
+     * https://docs.confluent.io/current/connect/references/restapi.html#put--connectors-(string-name)-resume
      *
      * @param connectorName Name of connector to resume.
      * @return Boolean true if success.
@@ -210,7 +266,7 @@ public Boolean deleteConnector(final String connectorName) {
 
     /**
      * Get a list of tasks currently running for the connector.
-     * https://docs.confluent.io/current/connect/restapi.html#get--connectors-(string-name)-tasks
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors-(string-name)-tasks
      *
      * @param connectorName Name of connector to retrieve tasks for.
      * @return Collection of details about each task.
@@ -221,7 +277,7 @@ public Collection<Task> getConnectorTasks(final String connectorName) {
 
     /**
      * Get a task’s status.
-     * https://docs.confluent.io/current/connect/restapi.html#get--connectors-(string-name)-tasks-(int-taskid)-status
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors-(string-name)-tasks-(int-taskid)-status
      *
      * @param connectorName Name of connector to retrieve tasks for.
      * @param taskId Id of task to get status for.
@@ -233,7 +289,7 @@ public TaskStatus getConnectorTaskStatus(final String connectorName, final int t
 
     /**
      * Restart an individual task.
-     * https://docs.confluent.io/current/connect/restapi.html#post--connectors-(string-name)-tasks-(int-taskid)-restart
+     * https://docs.confluent.io/current/connect/references/restapi.html#post--connectors-(string-name)-tasks-(int-taskid)-restart
      *
      * @param connectorName Name of connector to restart tasks for.
      * @param taskId Id of task to restart
@@ -245,7 +301,7 @@ public Boolean restartConnectorTask(final String connectorName, final int taskId
 
     /**
      * Return a list of connector plugins installed in the Kafka Connect cluster.
-     * https://docs.confluent.io/current/connect/restapi.html#get--connector-plugins-
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connector-plugins-
      *
      * @return Collection of available connector plugins.
      */
@@ -256,7 +312,7 @@ public Collection<ConnectorPlugin> 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
+     * https://docs.confluent.io/current/connect/references/restapi.html#put--connector-plugins-(string-name)-config-validate
      *
      * @param configDefinition Defines the configuration to validate.
      * @return Results of the validation.
@@ -286,6 +342,8 @@ private <T> T submitRequest(final Request<T> request) {
 
             try {
                 return request.parseResponse(responseStr);
+            } catch (final MismatchedInputException exception) {
+                throw new ResponseParseException(exception.getMessage(), exception);
             } catch (final IOException exception) {
                 throw new RuntimeException(exception.getMessage(), exception);
             }

src/main/java/org/sourcelab/kafka/connect/apiclient/exception/ResponseParseException.java

@@ -0,0 +1,34 @@
+/**
+ * Copyright 2018, 2019 SourceLab.org https://github.com/SourceLabOrg/kafka-connect-client
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit
+ * persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+ * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package org.sourcelab.kafka.connect.apiclient.exception;
+
+import com.fasterxml.jackson.databind.exc.MismatchedInputException;
+
+/**
+ * Thrown when the library is unable to properly parse the response from Kafka-Connect.
+ */
+public class ResponseParseException extends RuntimeException {
+    /**
+     * Constructor.
+     * @param message error msg
+     * @param exception underlying exception, if available.
+     */
+    public ResponseParseException(final String message, final MismatchedInputException exception) {
+        super(message, exception);
+    }
+}