Implement a few more end points

Author: Crim

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

@@ -5,14 +5,18 @@
 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.connector.DeleteConnector;
 import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorStatus;
 import org.sourcelab.kafka.connect.apiclient.request.get.connector.GetConnector;
 import org.sourcelab.kafka.connect.apiclient.request.get.connector.GetConnectorConfig;
 import org.sourcelab.kafka.connect.apiclient.request.get.connector.GetConnectorStatus;
 import org.sourcelab.kafka.connect.apiclient.request.get.connector.GetConnectors;
 import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorDefinition;
-import org.sourcelab.kafka.connect.apiclient.request.post.connector.PostConnectors;
+import org.sourcelab.kafka.connect.apiclient.request.post.connector.PostConnector;
+import org.sourcelab.kafka.connect.apiclient.request.post.connector.PostConnectorRestart;
 import org.sourcelab.kafka.connect.apiclient.request.put.connector.PutConnectorConfig;
+import org.sourcelab.kafka.connect.apiclient.request.put.connector.PutConnectorPause;
+import org.sourcelab.kafka.connect.apiclient.request.put.connector.PutConnectorResume;
 import org.sourcelab.kafka.connect.apiclient.rest.HttpClientRestClient;
 import org.sourcelab.kafka.connect.apiclient.rest.InvalidRequestException;
 import org.sourcelab.kafka.connect.apiclient.rest.RestClient;
@@ -79,8 +83,8 @@ private <T> T submitRequest(final Request<T> request) {
             }
 
             try {
-                return request.parseResponse(restResponse.getResponseStr());
-            } catch (IOException exception) {
+                return request.parseResponse(responseStr);
+            } catch (final IOException exception) {
                 throw new RuntimeException(exception.getMessage(), exception);
             }
         }
@@ -108,19 +112,78 @@ public Map<String, String> getConnectorConfig(final String connectorName) {
         return submitRequest(new GetConnectorConfig(connectorName));
     }
 
+    /**
+     * Get the status of specified connector by name.
+     * https://docs.confluent.io/current/connect/restapi.html#get--connectors-(string-name)-config
+     *
+     * @param connectorName Name of connector.
+     * @return Status details of the connector.
+     */
     public ConnectorStatus getConnectorStatus(final String connectorName) {
         return submitRequest(new GetConnectorStatus(connectorName));
     }
 
     // TODO Add return value
     public String addConnector(final ConnectorDefinition connectorDefinition) {
-        return submitRequest(new PostConnectors(connectorDefinition));
+        return submitRequest(new PostConnector(connectorDefinition));
     }
 
+    /**
+     * Update a connector's configuration.
+     * https://docs.confluent.io/current/connect/restapi.html#put--connectors-(string-name)-config
+     *
+     * @param connectorName Name of connector to update.
+     * @param config Configuration values to set.
+     * @return ConnectorDefinition describing the connectors configuration.
+     */
     public ConnectorDefinition updateConnectorConfig(final String connectorName, final Map<String, String> config) {
         return submitRequest(new PutConnectorConfig(connectorName, config));
     }
 
+    /**
+     * Restart a connector.
+     * https://docs.confluent.io/current/connect/restapi.html#post--connectors-(string-name)-restart
+     *
+     * @param connectorName Name of connector to restart.
+     * @return Boolean true if success.
+     */
+    public Boolean restartConnector(final String connectorName) {
+        return submitRequest(new PostConnectorRestart(connectorName));
+    }
+
+    /**
+     * Pause a connector.
+     * https://docs.confluent.io/current/connect/restapi.html#put--connectors-(string-name)-pause
+     *
+     * @param connectorName Name of connector to pause.
+     * @return Boolean true if success.
+     */
+    public Boolean pauseConnector(final String connectorName) {
+        return submitRequest(new PutConnectorPause(connectorName));
+    }
+
+    /**
+     * Resume a connector.
+     * https://docs.confluent.io/current/connect/restapi.html#put--connectors-(string-name)-resume
+     *
+     * @param connectorName Name of connector to resume.
+     * @return Boolean true if success.
+     */
+    public Boolean resumeConnector(final String connectorName) {
+        return submitRequest(new PutConnectorResume(connectorName));
+    }
+
+    /**
+     * Resume a connector.
+     * https://docs.confluent.io/current/connect/restapi.html#put--connectors-(string-name)-resume
+     *
+     * @param connectorName Name of connector to resume.
+     * @return Boolean true if success.
+     */
+    public Boolean deleteConnector(final String connectorName) {
+        return submitRequest(new DeleteConnector(connectorName));
+    }
+
     /**
      * package protected for access in tests.
      * @return Rest Client.

src/main/java/org/sourcelab/kafka/connect/apiclient/request/RequestMethod.java

@@ -1,7 +1,8 @@
 package org.sourcelab.kafka.connect.apiclient.request;
 
 public enum RequestMethod {
+    DELETE,
     GET,
     POST,
-    PUT;
+    PUT
 }

src/main/java/org/sourcelab/kafka/connect/apiclient/request/delete/DeleteRequest.java

@@ -0,0 +1,11 @@
+package org.sourcelab.kafka.connect.apiclient.request.delete;
+
+import org.sourcelab.kafka.connect.apiclient.request.Request;
+import org.sourcelab.kafka.connect.apiclient.request.RequestMethod;
+
+public interface DeleteRequest<T> extends Request<T> {
+    @Override
+    default RequestMethod getRequestMethod() {
+        return RequestMethod.DELETE;
+    }
+}

src/main/java/org/sourcelab/kafka/connect/apiclient/request/delete/connector/DeleteConnector.java

@@ -0,0 +1,29 @@
+package org.sourcelab.kafka.connect.apiclient.request.delete.connector;
+
+import com.sun.xml.internal.rngom.util.Uri;
+import org.sourcelab.kafka.connect.apiclient.request.delete.DeleteRequest;
+
+import java.io.IOException;
+
+public class DeleteConnector implements DeleteRequest<Boolean> {
+    private final String name;
+
+    public DeleteConnector(final String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String getApiEndpoint() {
+        return "/connectors/" + Uri.escapeDisallowedChars(name);
+    }
+
+    @Override
+    public Object getRequestBody() {
+        return "";
+    }
+
+    @Override
+    public Boolean parseResponse(final String responseStr) throws IOException {
+        return true;
+    }
+}

…quest/post/connector/PostConnectors.java …equest/post/connector/PostConnector.javasrc/main/java/org/sourcelab/kafka/connect/apiclient/request/post/connector/PostConnectors.java renamed to src/main/java/org/sourcelab/kafka/connect/apiclient/request/post/connector/PostConnector.java src/main/java/org/sourcelab/kafka/connect/apiclient/request/post/connector/PostConnectors.java renamed to src/main/java/org/sourcelab/kafka/connect/apiclient/request/post/connector/PostConnector.java

@@ -5,10 +5,10 @@
 
 import java.io.IOException;
 
-public class PostConnectors implements PostRequest<String> {
+public class PostConnector implements PostRequest<String> {
     private final ConnectorDefinition connectorDefinition;
 
-    public PostConnectors(final ConnectorDefinition connectorDefinition) {
+    public PostConnector(final ConnectorDefinition connectorDefinition) {
         if (connectorDefinition == null) {
             throw new NullPointerException("ConnectorDefinition parameter cannot be a null reference!");
         }

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

@@ -0,0 +1,29 @@
+package org.sourcelab.kafka.connect.apiclient.request.post.connector;
+
+import com.sun.xml.internal.rngom.util.Uri;
+import org.sourcelab.kafka.connect.apiclient.request.post.PostRequest;
+
+import java.io.IOException;
+
+public class PostConnectorRestart implements PostRequest<Boolean> {
+    private final String name;
+
+    public PostConnectorRestart(final String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String getApiEndpoint() {
+        return "/connectors/" + Uri.escapeDisallowedChars(name) + "/restart";
+    }
+
+    @Override
+    public Object getRequestBody() {
+        return "";
+    }
+
+    @Override
+    public Boolean parseResponse(final String responseStr) throws IOException {
+        return true;
+    }
+}

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

@@ -0,0 +1,29 @@
+package org.sourcelab.kafka.connect.apiclient.request.put.connector;
+
+import com.sun.xml.internal.rngom.util.Uri;
+import org.sourcelab.kafka.connect.apiclient.request.put.PutRequest;
+
+import java.io.IOException;
+
+public class PutConnectorPause implements PutRequest<Boolean> {
+    private final String name;
+
+    public PutConnectorPause(final String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String getApiEndpoint() {
+        return "/connectors/" + Uri.escapeDisallowedChars(name) + "/pause";
+    }
+
+    @Override
+    public Object getRequestBody() {
+        return null;
+    }
+
+    @Override
+    public Boolean parseResponse(final String responseStr) throws IOException {
+        return true;
+    }
+}

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

@@ -0,0 +1,29 @@
+package org.sourcelab.kafka.connect.apiclient.request.put.connector;
+
+import com.sun.xml.internal.rngom.util.Uri;
+import org.sourcelab.kafka.connect.apiclient.request.put.PutRequest;
+
+import java.io.IOException;
+
+public class PutConnectorResume implements PutRequest<Boolean> {
+    private final String name;
+
+    public PutConnectorResume(final String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String getApiEndpoint() {
+        return "/connectors/" + Uri.escapeDisallowedChars(name) + "/resume";
+    }
+
+    @Override
+    public Object getRequestBody() {
+        return null;
+    }
+
+    @Override
+    public Boolean parseResponse(final String responseStr) throws IOException {
+        return true;
+    }
+}

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

@@ -9,6 +9,7 @@
 import org.apache.http.client.ResponseHandler;
 import org.apache.http.client.config.RequestConfig;
 import org.apache.http.client.entity.UrlEncodedFormEntity;
+import org.apache.http.client.methods.HttpDelete;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpPut;
@@ -157,6 +158,8 @@ public RestResponse submitRequest(final Request request) throws RestException {
                     return submitPostRequest(url, request.getRequestBody(), responseHandler);
                 case PUT:
                     return submitPutRequest(url, request.getRequestBody(), responseHandler);
+                case DELETE:
+                    return submitDeleteRequest(url, request.getRequestBody(), responseHandler);
                 default:
                     throw new IllegalArgumentException("Unknown Request Method: " + request.getRequestMethod());
             }
@@ -294,6 +297,50 @@ private <T> T submitPutRequest(final String url, final Object requestBody, final
         return null;
     }
 
+    /**
+     * Internal DELETE method.
+     * @param url Url to DELETE to.
+     * @param requestBody POST entity include in the request body
+     * @param responseHandler The response Handler to use to parse the response
+     * @param <T> The type that ResponseHandler returns.
+     * @return Parsed response.
+     */
+    private <T> T submitDeleteRequest(final String url, final Object requestBody, final ResponseHandler<T> responseHandler) throws IOException {
+        try {
+            // Construct URI including our request parameters.
+            final URIBuilder uriBuilder = new URIBuilder(url)
+                .setCharset(StandardCharsets.UTF_8);
+
+            final HttpDelete delete = new HttpDelete(url);
+
+            // Add Accept header.
+            delete.addHeader(new BasicHeader("Accept", "application/json"));
+
+            // Conditionally add content-type header?
+            delete.addHeader(new BasicHeader("Content-Type", "application/json"));
+
+            // Define required auth params
+            final List<NameValuePair> params = new ArrayList<>();
+
+            // Convert to Json
+            final String jsonPayloadStr = JacksonFactory.newInstance().writeValueAsString(requestBody);
+
+            logger.info("Executing request {} with {}", delete.getRequestLine(), jsonPayloadStr);
+
+            // Execute and return
+            return httpClient.execute(delete, responseHandler);
+        } catch (ClientProtocolException e) {
+            e.printStackTrace();
+        } catch (IOException e) {
+            // Typically this is a parse error.
+            e.printStackTrace();
+        } catch (URISyntaxException e) {
+            // Bad URI building
+            e.printStackTrace();
+        }
+        return null;
+    }
+
     /**
      * Internal helper method for generating URLs w/ the appropriate API host and API version.
      * @param endPoint The end point you want to hit.

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

@@ -88,11 +88,49 @@ public void testAddConnector() {
      */
     @Test
     public void testUpdateConnectorConfig() {
+        final String connectorName = "My Test Connector";
+
         final Map<String, String> config = new HashMap<>();
         config.put("connector.class", "org.apache.kafka.connect.tools.MockConnector");
         config.put("tasks.max", "10");
         config.put("topics", "test-topic");
 
-        logger.info("Result: {}", apiClient.updateConnectorConfig("My Test Connector", config));
+        logger.info("Result: {}", apiClient.updateConnectorConfig(connectorName, config));
+    }
+
+    /**
+     * Test restarting a connector.
+     */
+    @Test
+    public void testRestartConnector() {
+        final String connectorName = "My Test Connector";
+        logger.info("Result: {}", apiClient.restartConnector(connectorName));
+    }
+
+    /**
+     * Test pausing a connector.
+     */
+    @Test
+    public void testPauseConnector() {
+        final String connectorName = "My Test Connector";
+        logger.info("Result: {}", apiClient.pauseConnector(connectorName));
+    }
+
+    /**
+     * Test pausing a connector.
+     */
+    @Test
+    public void testResumeConnector() {
+        final String connectorName = "My Test Connector";
+        logger.info("Result: {}", apiClient.resumeConnector(connectorName));
+    }
+
+    /**
+     * Test pausing a connector.
+     */
+    @Test
+    public void testDeleteConnector() {
+        final String connectorName = "My Test Connector";
+        logger.info("Result: {}", apiClient.deleteConnector(connectorName));
     }
 }