add support for new connector restart request parameters

Author: Crim

CHANGELOG.md

@@ -2,10 +2,20 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
-## 3.1.4 (12/10/2021)
-- Remove Guava as a dependency.
-- Update Log4j dependency from 2.14.1 to 2.15.0 for [CVE-2021-44228](https://nvd.nist.gov/vuln/detail/CVE-2021-44228).
+## 4.0.0 (12/10/2021)
 
+**Note:** Major release version updated purely out of abundance of caution.
+
+Internal dependency on Google Guava was removed.  Functionally nothing about this library's
+API has changed.  If you were making use of Google Guava via a transitive dependency provided by this
+library, then you'll need to update your project to include it directly.
+
+### New Features
+- Update [/connector/\<name\>/restart](https://docs.confluent.io/platform/current/connect/references/restapi.html#post--connectors-(string-name)-restart) request to support the `includeTasks` and `onlyFailed` parameters from Kafka-Connect 3.0.0. 
+
+#### Internal Dependency Updates
+- Removed Google Guava as a dependency.
+- com.fasterxml.jackson.core from 2.12.2 -> 2.13.0.
 
 ## 3.1.3 (08/11/2021)
 - [Issue-55](https://github.com/SourceLabOrg/kafka-connect-client/issues/55) Create new HttpContext for every request.

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>3.1.3</version>
+    <version>3.1.4</version>
 </dependency>
 ```
 

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>org.sourcelab</groupId>
     <artifactId>kafka-connect-client</artifactId>
-    <version>3.1.3</version>
+    <version>4.0.0</version>
     <packaging>jar</packaging>
 
     <!-- Require Maven 3.3.9 -->
@@ -46,14 +46,11 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 
-        <!-- guava version -->
-        <guava.version>30.1.1-jre</guava.version>
-
         <!-- Http Components version -->
         <http-components.version>4.5.13</http-components.version>
 
         <!-- Jackson version -->
-        <jackson.version>2.12.2</jackson.version>
+        <jackson.version>2.13.0</jackson.version>
 
         <!-- Define which version of junit you'll be running -->
         <junit.version>4.13.1</junit.version>
@@ -64,7 +61,7 @@
         <checkstyle.version>8.32</checkstyle.version>
 
         <!-- Log4J Version -->
-        <log4j2.version>2.14.1</log4j2.version>
+        <log4j2.version>2.15.0</log4j2.version>
         <slf4j.version>1.7.32</slf4j.version>
 
         <!-- test toggling -->

src/main/java/org/sourcelab/kafka/connect/apiclient/request/post/PostConnectorRestart.java

@@ -20,6 +20,10 @@
 import org.sourcelab.kafka.connect.apiclient.util.UrlEscapingUtil;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 /**
@@ -28,6 +32,8 @@
 public final class PostConnectorRestart implements PostRequest<Boolean> {
     private final String connectorName;
 
+    private Map<String, Boolean> options = new HashMap<>();
+
     /**
      * Constructor.
      * @param connectorName Name of connector to restart
@@ -37,9 +43,46 @@ public PostConnectorRestart(final String connectorName) {
         this.connectorName = connectorName;
     }
 
+    /**
+     * Only available from Kafka Connect version 3.0.0 and up.
+     * @param includeTasks Specifies whether to restart the connector instance and task instances (includeTasks=true`) or just the connector instance (includeTasks=false).
+     * @return self reference for method chaining.
+     */
+    public PostConnectorRestart withIncludeTasks(final boolean includeTasks)
+    {
+        options.put("includeTasks", includeTasks);
+        return this;
+    }
+
+    /**
+     * Only available from Kafka Connect version 3.0.0 and up.
+     * @param onlyFailed specifies whether to restart just the instances with a FAILED status (onlyFailed=true) or all instances (onlyFailed=false).
+     * @return self reference for method chaining.
+     */
+    public PostConnectorRestart withOnlyFailed(final boolean onlyFailed)
+    {
+        options.put("onlyFailed", onlyFailed);
+        return this;
+    }
+
     @Override
     public String getApiEndpoint() {
-        return "/connectors/" + UrlEscapingUtil.escapePath(connectorName) + "/restart";
+        // Define base URL
+        String url = "/connectors/" + UrlEscapingUtil.escapePath(connectorName) + "/restart";
+
+        // Optionally add additional request parameters if explicitly defined.
+        final List<String> params = new ArrayList<>();
+        for (final Map.Entry<String, Boolean> option : options.entrySet()) {
+            // skip null
+            if (option.getValue() == null) {
+                continue;
+            }
+            params.add(option.getKey() + "=" + (option.getValue() ? "true" : "false"));
+        }
+        if (params.size() > 0) {
+            url = url + "?" + String.join("&", params);
+        }
+        return url;
     }
 
     @Override
@@ -49,6 +92,7 @@ public Object getRequestBody() {
 
     @Override
     public Boolean parseResponse(final String responseStr) throws IOException {
+        // Note: this doesn't currently support 202 responses which would normally contain a response body :/
         return true;
     }
 }

src/test/java/org/sourcelab/kafka/connect/apiclient/request/post/PostConnectorRestartTest.java

@@ -0,0 +1,98 @@
+package org.sourcelab.kafka.connect.apiclient.request.post;
+
+import org.junit.Test;
+import org.sourcelab.kafka.connect.apiclient.request.AbstractRequestTest;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ *
+ */
+public class PostConnectorRestartTest extends AbstractRequestTest {
+    @Override
+    public void testParseResponse() throws Exception {
+
+    }
+
+    /**
+     * With no optional parameters.
+     */
+    @Override
+    public void getApiEndpoint() {
+        final String inputName = "My Test Connector";
+        final String expectedUrl = "/connectors/My%20Test%20Connector/restart";
+        final String result = new PostConnectorRestart(inputName).getApiEndpoint();
+        assertEquals("Unexpected URL returned!", expectedUrl, result);
+    }
+
+    /**
+     * Verify with various optional parameters specified.
+     */
+    @Test
+    public void getApiEndpoint_includeTasks_true() {
+        final String inputName = "My Test Connector";
+        final String expectedUrl = "/connectors/My%20Test%20Connector/restart?includeTasks=true";
+        final String result = new PostConnectorRestart(inputName)
+            .withIncludeTasks(true)
+            .getApiEndpoint();
+
+        assertEquals("Unexpected URL returned!", expectedUrl, result);
+    }
+
+    /**
+     * Verify with various optional parameters specified.
+     */
+    @Test
+    public void getApiEndpoint_includeTasks_false() {
+        final String inputName = "My Test Connector";
+        final String expectedUrl = "/connectors/My%20Test%20Connector/restart?includeTasks=false";
+        final String result = new PostConnectorRestart(inputName)
+            .withIncludeTasks(false)
+            .getApiEndpoint();
+
+        assertEquals("Unexpected URL returned!", expectedUrl, result);
+    }
+
+    /**
+     * Verify with various optional parameters specified.
+     */
+    @Test
+    public void getApiEndpoint_onlyFailed_true() {
+        final String inputName = "My Test Connector";
+        final String expectedUrl = "/connectors/My%20Test%20Connector/restart?onlyFailed=true";
+        final String result = new PostConnectorRestart(inputName)
+            .withOnlyFailed(true)
+            .getApiEndpoint();
+
+        assertEquals("Unexpected URL returned!", expectedUrl, result);
+    }
+
+    /**
+     * Verify with various optional parameters specified.
+     */
+    @Test
+    public void getApiEndpoint_onlyFailed_false() {
+        final String inputName = "My Test Connector";
+        final String expectedUrl = "/connectors/My%20Test%20Connector/restart?onlyFailed=false";
+        final String result = new PostConnectorRestart(inputName)
+            .withOnlyFailed(false)
+            .getApiEndpoint();
+
+        assertEquals("Unexpected URL returned!", expectedUrl, result);
+    }
+
+    /**
+     * Verify with various optional parameters specified.
+     */
+    @Test
+    public void getApiEndpoint_onlyFailed_includeTasks_true() {
+        final String inputName = "My Test Connector";
+        final String expectedUrl = "/connectors/My%20Test%20Connector/restart?includeTasks=true&onlyFailed=true";
+        final String result = new PostConnectorRestart(inputName)
+            .withOnlyFailed(true)
+            .withIncludeTasks(true)
+            .getApiEndpoint();
+
+        assertEquals("Unexpected URL returned!", expectedUrl, result);
+    }
+}