Merge pull request #43 from SourceLabOrg/sp/newEndPoints

add support for new kafka connect reset end points

Author: Crim

CHANGELOG.md

@@ -2,6 +2,18 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## 3.1.0 (05/10/2020)
+
+### New Features
+- Add support for `/connectors/<connector-name>/topics` and `/connectors/<connector-name>/topics/reset` endpoints 
+added in Kafka 2.5.0 via `getConnectorTopics()` and `resetConnectorTopics()` methods on KafkaConnectClient.
+
+#### Internal Dependency Updates
+- com.google.guava:guava from 28.2-jre -> 29.0-jre
+- org.apache.httpcomponents from 4.5.11 -> 4.5.12
+- com.fasterxml.jackson.core from 2.10.2 -> 2.10.4
+- Checkstyle plugin from 8.24 -> 8.32
+
 ## 3.0.0 (03/20/2020)
 
 #### Possible Breaking Dependency Change

build/checkstyle-v1.5.xml

@@ -156,11 +156,11 @@
         <module name="JavadocMethod">
             <property name="scope" value="public"/>
             <property name="allowMissingParamTags" value="true"/>
-            <property name="allowMissingThrowsTags" value="true"/>
             <property name="allowMissingReturnTag" value="true"/>
-            <property name="minLineCount" value="2"/>
             <property name="allowedAnnotations" value="Override, Test"/>
-            <property name="allowThrowsTagsForSubclasses" value="true"/>
+        </module>
+        <module name="MissingJavadocMethod">
+            <property name="minLineCount" value="2"/>
         </module>
         <module name="JavadocStyle">
             <property name="scope" value="public"/>

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>org.sourcelab</groupId>
     <artifactId>kafka-connect-client</artifactId>
-    <version>3.0.0</version>
+    <version>3.1.0</version>
     <packaging>jar</packaging>
 
     <!-- Require Maven 3.3.9 -->
@@ -47,24 +47,24 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 
         <!-- guava version -->
-        <guava.version>28.2-jre</guava.version>
+        <guava.version>29.0-jre</guava.version>
 
         <!-- Http Components version -->
-        <http-components.version>4.5.11</http-components.version>
+        <http-components.version>4.5.12</http-components.version>
 
         <!-- Jackson version -->
-        <jackson.version>2.10.2</jackson.version>
+        <jackson.version>2.10.4</jackson.version>
 
         <!-- Define which version of junit you'll be running -->
         <junit.version>4.12</junit.version>
 
         <!-- Specify which Checkstyle ruleset to use -->
         <checkstyle.ruleset>build/checkstyle-v1.5.xml</checkstyle.ruleset>
-        <checkstyle.plugin.version>3.0.0</checkstyle.plugin.version>
-        <checkstyle.version>8.24</checkstyle.version>
+        <checkstyle.plugin.version>3.1.1</checkstyle.plugin.version>
+        <checkstyle.version>8.32</checkstyle.version>
 
         <!-- Log4J Version -->
-        <log4j2.version>2.12.1</log4j2.version>
+        <log4j2.version>2.13.2</log4j2.version>
         <slf4j.version>1.7.30</slf4j.version>
 
         <!-- test toggling -->
@@ -156,7 +156,7 @@
         <dependency>
             <groupId>org.eclipse.jetty</groupId>
             <artifactId>jetty-server</artifactId>
-            <version>9.4.27.v20200227</version>
+            <version>9.4.28.v20200408</version>
             <scope>test</scope>
         </dependency>
 
@@ -187,7 +187,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.6.1</version>
+                <version>3.8.1</version>
                 <configuration>
                     <source>1.8</source>
                     <target>1.8</target>
@@ -343,7 +343,7 @@
                     <plugin>
                         <groupId>org.apache.maven.plugins</groupId>
                         <artifactId>maven-source-plugin</artifactId>
-                        <version>3.0.1</version>
+                        <version>3.2.1</version>
                         <executions>
                             <execution>
                                 <id>attach-sources</id>
@@ -358,7 +358,7 @@
                     <plugin>
                         <groupId>org.apache.maven.plugins</groupId>
                         <artifactId>maven-javadoc-plugin</artifactId>
-                        <version>3.0.1</version>
+                        <version>3.2.0</version>
                         <executions>
                             <execution>
                                 <id>attach-javadocs</id>

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

@@ -32,6 +32,7 @@
 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.ConnectorTopics;
 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;
@@ -45,6 +46,7 @@
 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.GetConnectorTopics;
 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;
@@ -56,6 +58,7 @@
 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.request.put.PutConnectorTopicsReset;
 import org.sourcelab.kafka.connect.apiclient.rest.HttpClientRestClient;
 import org.sourcelab.kafka.connect.apiclient.rest.RestClient;
 import org.sourcelab.kafka.connect.apiclient.rest.RestResponse;
@@ -112,6 +115,7 @@ public KafkaConnectClient(final Configuration configuration, final RestClient re
 
     /**
      * Retrieve details about the Kafka-Connect service itself.
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--
      * @return ConnectServerVersion
      */
     public ConnectServerVersion getConnectServerVersion() {
@@ -197,6 +201,30 @@ public ConnectorStatus getConnectorStatus(final String connectorName) {
         return submitRequest(new GetConnectorStatus(connectorName));
     }
 
+    /**
+     * Get the set of topics that a specific connector is using since the connector was created or since a request
+     * to reset its set of active topics was issued.
+     * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors-(string-name)-topics
+     *
+     * Requires Kafka-Connect 2.5.0+
+     *
+     * @param connectorName Name of connector.
+     */
+    public ConnectorTopics getConnectorTopics(final String connectorName) {
+        return submitRequest(new GetConnectorTopics(connectorName));
+    }
+
+    /**
+     * Send a request to empty the set of active topics of a connector.
+     * https://docs.confluent.io/current/connect/references/restapi.html#put--connectors-(string-name)-topics-reset
+     * Requires Kafka-Connect 2.5.0+
+     *
+     * @param connectorName Name of connector.
+     */
+    public boolean resetConnectorTopics(final String connectorName) {
+        return submitRequest(new PutConnectorTopicsReset(connectorName));
+    }
+
     /**
      * Create a new connector, returning the current connector info if successful.
      * https://docs.confluent.io/current/connect/references/restapi.html#post--connectors

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

@@ -0,0 +1,122 @@
+/**
+ * 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.request.dto;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import com.fasterxml.jackson.databind.node.JsonNodeType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * Represents result from /connectors/[topic-name]/topics REST end point.
+ */
+@JsonDeserialize(using = ConnectorTopics.Deserializer.class)
+public class ConnectorTopics {
+    private final String name;
+    private final List<String> topics;
+
+    /**
+     * Constructor.
+     * @param name Name of the connector.
+     * @param topics List of topics.
+     */
+    public ConnectorTopics(final String name, final List<String> topics) {
+        this.name = Objects.requireNonNull(name);
+        Objects.requireNonNull(topics);
+        this.topics = Collections.unmodifiableList(topics.stream().sorted().collect(Collectors.toList()));
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public List<String> getTopics() {
+        return topics;
+    }
+
+    @Override
+    public String toString() {
+        return "ConnectorTopics{"
+            + "name='" + name + '\''
+            + ", topics=" + topics
+            + '}';
+    }
+
+    /**
+     * Deserializer for ConnectorTopics.
+     */
+    public static class Deserializer extends StdDeserializer<ConnectorTopics> {
+
+        /**
+         * Constructor.
+         */
+        public Deserializer() {
+            this(null);
+        }
+
+        /**
+         * Constructor.
+         */
+        public Deserializer(final Class<?> vc) {
+            super(vc);
+        }
+
+        @Override
+        public ConnectorTopics deserialize(final JsonParser jsonParser, final DeserializationContext ctxt) throws IOException, JsonProcessingException {
+            final JsonNode node = jsonParser.getCodec().readTree(jsonParser);
+            final Iterator<Map.Entry<String, JsonNode>> children = node.fields();
+
+            while (children.hasNext()) {
+                final Map.Entry<String, JsonNode> child = children.next();
+                final String name = child.getKey();
+                final JsonNode childNode = child.getValue();
+
+                // Skip unknown entries
+                if (JsonNodeType.OBJECT != childNode.getNodeType()) {
+                    continue;
+                }
+
+                // If has no topics key
+                if (!childNode.has("topics") || JsonNodeType.ARRAY != childNode.get("topics").getNodeType()) {
+                    // Skip?
+                    continue;
+                }
+                final List<String> topicNames = new ArrayList<>();
+                if (!childNode.get("topics").isEmpty()) {
+                    // Parse topic name values out
+                    childNode.get("topics").forEach((entry) -> topicNames.add(entry.textValue()));
+                }
+                return new ConnectorTopics(name, topicNames);
+            }
+            throw new JsonParseException(jsonParser, "Unable to parse response JSON");
+        }
+    }
+}

src/main/java/org/sourcelab/kafka/connect/apiclient/request/get/GetConnectorTopics.java

@@ -0,0 +1,57 @@
+/**
+ * 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.request.get;
+
+import org.sourcelab.kafka.connect.apiclient.request.JacksonFactory;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorTopics;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static com.google.common.net.UrlEscapers.urlPathSegmentEscaper;
+
+/**
+ * Returns a list of connector topic names.
+ * There is no defined order in which the topics are returned and consecutive calls may return the same topic names
+ * but in different order. This request is independent of whether a connector is running, and will return an empty set
+ * of topics, both for connectors that don’t have active topics as well as non-existent connectors.
+ *
+ * https://docs.confluent.io/current/connect/references/restapi.html#get--connectors-(string-name)-topics
+ */
+public class GetConnectorTopics implements GetRequest<ConnectorTopics> {
+    private final String connectorName;
+
+    /**
+     * Constructor.
+     * @param connectorName Name of connector.
+     */
+    public GetConnectorTopics(final String connectorName) {
+        Objects.requireNonNull(connectorName);
+        this.connectorName = connectorName;
+    }
+
+    @Override
+    public String getApiEndpoint() {
+        return "/connectors/" + urlPathSegmentEscaper().escape(connectorName) + "/topics";
+    }
+
+    @Override
+    public ConnectorTopics parseResponse(final String responseStr) throws IOException {
+        return JacksonFactory.newInstance().readValue(responseStr, ConnectorTopics.class);
+    }
+}