feat: add DROP CONNECTOR functionality (#3245)

ksql-cli/src/main/java/io/confluent/ksql/cli/console/Console.java

@@ -29,6 +29,7 @@
 import io.confluent.ksql.cli.console.table.builder.CommandStatusTableBuilder;
 import io.confluent.ksql.cli.console.table.builder.ConnectorInfoTableBuilder;
 import io.confluent.ksql.cli.console.table.builder.ConnectorListTableBuilder;
+import io.confluent.ksql.cli.console.table.builder.DropConnectorTableBuilder;
 import io.confluent.ksql.cli.console.table.builder.ErrorEntityTableBuilder;
 import io.confluent.ksql.cli.console.table.builder.ExecutionPlanTableBuilder;
 import io.confluent.ksql.cli.console.table.builder.FunctionNameListTableBuilder;
@@ -45,6 +46,7 @@
 import io.confluent.ksql.rest.entity.ConnectorDescription;
 import io.confluent.ksql.rest.entity.ConnectorList;
 import io.confluent.ksql.rest.entity.CreateConnectorEntity;
+import io.confluent.ksql.rest.entity.DropConnectorEntity;
 import io.confluent.ksql.rest.entity.ErrorEntity;
 import io.confluent.ksql.rest.entity.ExecutionPlan;
 import io.confluent.ksql.rest.entity.FieldInfo;
@@ -149,6 +151,8 @@ public class Console implements Closeable {
               Console::printFunctionDescription)
           .put(CreateConnectorEntity.class,
               tablePrinter(CreateConnectorEntity.class, ConnectorInfoTableBuilder::new))
+          .put(DropConnectorEntity.class,
+              tablePrinter(DropConnectorEntity.class, DropConnectorTableBuilder::new))
           .put(ConnectorList.class,
               tablePrinter(ConnectorList.class, ConnectorListTableBuilder::new))
           .put(ConnectorDescription.class,

ksql-cli/src/main/java/io/confluent/ksql/cli/console/table/builder/DropConnectorTableBuilder.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"; you may not use
+ * this file except in compliance with the License. You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.cli.console.table.builder;
+
+import io.confluent.ksql.cli.console.table.Table;
+import io.confluent.ksql.rest.entity.DropConnectorEntity;
+
+public class DropConnectorTableBuilder implements TableBuilder<DropConnectorEntity> {
+
+  @Override
+  public Table buildTable(final DropConnectorEntity entity) {
+    return new Table.Builder()
+        .withColumnHeaders("Message")
+        .withRow("Dropped connector \"" + entity.getConnectorName() + '"')
+        .build();
+  }
+}

ksql-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.java

@@ -41,6 +41,7 @@
 import io.confluent.ksql.rest.entity.CommandStatusEntity;
 import io.confluent.ksql.rest.entity.ConnectorDescription;
 import io.confluent.ksql.rest.entity.ConnectorList;
+import io.confluent.ksql.rest.entity.DropConnectorEntity;
 import io.confluent.ksql.rest.entity.EntityQueryId;
 import io.confluent.ksql.rest.entity.ErrorEntity;
 import io.confluent.ksql.rest.entity.ExecutionPlan;
@@ -1039,6 +1040,37 @@ public void shouldPrintWarnings() throws IOException {
     }
   }
 
+  @Test
+  public void shouldPrintDropConnector() throws IOException {
+    // Given:
+    final KsqlEntity entity = new DropConnectorEntity("statementText", "connectorName");
+
+    // When:
+    console.printKsqlEntityList(ImmutableList.of(entity));
+
+    // Then:
+    final String output = terminal.getOutputString();
+    if (console.getOutputFormat() == OutputFormat.TABULAR) {
+      assertThat(
+          output,
+          is("\n"
+              + " Message                           \n"
+              + "-----------------------------------\n"
+              + " Dropped connector \"connectorName\" \n"
+              + "-----------------------------------\n")
+      );
+    } else {
+      assertThat(
+          output,
+          is("[ {\n"
+              + "  \"statementText\" : \"statementText\",\n"
+              + "  \"connectorName\" : \"connectorName\",\n"
+              + "  \"warnings\" : [ ]\n"
+              + "} ]\n")
+      );
+    }
+  }
+
   @Test
   public void shouldPrintErrorEntityLongNonJson() throws IOException {
     // Given:

ksql-engine/src/main/java/io/confluent/ksql/services/ConnectClient.java

@@ -58,6 +58,13 @@ public interface ConnectClient {
    */
   ConnectResponse<ConnectorStateInfo> status(String connector);
 
+  /**
+   * Delete the {@code connector}.
+   *
+   * @param connector the connector name
+   */
+  ConnectResponse<String> delete(String connector);
+
   /**
    * An optionally successful response. Either contains a value of type
    * {@code <T>} or an error, which is the string representation of the
@@ -68,11 +75,11 @@ class ConnectResponse<T> {
     private final Optional<String> error;
     private final int httpCode;
 
-    public static <T> ConnectResponse<T> of(final T datum, final int code) {
+    public static <T> ConnectResponse<T> success(final T datum, final int code) {
       return new ConnectResponse<>(datum, null, code);
     }
 
-    public static <T> ConnectResponse<T> of(final String error, final int code) {
+    public static <T> ConnectResponse<T> failure(final String error, final int code) {
       return new ConnectResponse<>(null, error, code);
     }
 

ksql-engine/src/main/java/io/confluent/ksql/services/DefaultConnectClient.java

@@ -32,6 +32,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Function;
+import org.apache.http.HttpEntity;
 import org.apache.http.HttpStatus;
 import org.apache.http.client.ResponseHandler;
 import org.apache.http.client.fluent.Request;
@@ -176,6 +177,29 @@ public ConnectResponse<ConnectorInfo> describe(final String connector) {
     }
   }
 
+  @Override
+  public ConnectResponse<String> delete(final String connector) {
+    try {
+      LOG.debug("Issuing request to Kafka Connect at URI {} to delete {}",
+          connectUri, connector);
+
+      final ConnectResponse<String> connectResponse = withRetries(() -> Request
+          .Delete(connectUri.resolve(String.format("%s/%s", CONNECTORS, connector)))
+          .socketTimeout(DEFAULT_TIMEOUT_MS)
+          .connectTimeout(DEFAULT_TIMEOUT_MS)
+          .execute()
+          .handleResponse(
+              createHandler(HttpStatus.SC_NO_CONTENT, Object.class, foo -> connector)));
+
+      connectResponse.error()
+          .ifPresent(error -> LOG.warn("Could not delete connector: {}.", error));
+
+      return connectResponse;
+    } catch (final Exception e) {
+      throw new KsqlServerException(e);
+    }
+  }
+
   @SuppressWarnings("unchecked")
   private static <T> ConnectResponse<T> withRetries(final Callable<ConnectResponse<T>> action) {
     try {
@@ -211,14 +235,17 @@ private static <T, C> ResponseHandler<ConnectResponse<T>> createHandler(
       final int code = httpResponse.getStatusLine().getStatusCode();
       if (httpResponse.getStatusLine().getStatusCode() != expectedStatus) {
         final String entity = EntityUtils.toString(httpResponse.getEntity());
-        return ConnectResponse.of(entity, code);
+        return ConnectResponse.failure(entity, code);
       }
 
-      final T info = cast.apply(MAPPER.readValue(
-          httpResponse.getEntity().getContent(),
-          entityClass));
+      final HttpEntity entity = httpResponse.getEntity();
+      final T data = cast.apply(
+          entity == null
+              ? null
+              : MAPPER.readValue(entity.getContent(), entityClass)
+      );
 
-      return ConnectResponse.of(info, code);
+      return ConnectResponse.success(data, code);
     };
   }
 }

ksql-engine/src/main/java/io/confluent/ksql/services/SandboxConnectClient.java

@@ -34,13 +34,15 @@ private SandboxConnectClient() { }
   public static ConnectClient createProxy() {
     return LimitedProxyBuilder.forClass(ConnectClient.class)
         .swallow("create", methodParams(String.class, Map.class),
-            ConnectResponse.of("sandbox", HttpStatus.SC_INTERNAL_SERVER_ERROR))
+            ConnectResponse.failure("sandbox", HttpStatus.SC_INTERNAL_SERVER_ERROR))
         .swallow("describe", methodParams(String.class),
-            ConnectResponse.of("sandbox", HttpStatus.SC_INTERNAL_SERVER_ERROR))
+            ConnectResponse.failure("sandbox", HttpStatus.SC_INTERNAL_SERVER_ERROR))
         .swallow("connectors", methodParams(),
-            ConnectResponse.of(ImmutableList.of(), HttpStatus.SC_OK))
+            ConnectResponse.success(ImmutableList.of(), HttpStatus.SC_OK))
         .swallow("status", methodParams(String.class),
-            ConnectResponse.of("sandbox", HttpStatus.SC_INTERNAL_SERVER_ERROR))
+            ConnectResponse.failure("sandbox", HttpStatus.SC_INTERNAL_SERVER_ERROR))
+        .swallow("delete", methodParams(String.class),
+            ConnectResponse.success("sandbox", HttpStatus.SC_NO_CONTENT))
         .build();
   }
 }

ksql-engine/src/test/java/io/confluent/ksql/connect/ConnectConfigServiceTest.java

@@ -228,14 +228,14 @@ public void shouldIgnoreConnectClientFailureToDescribe() throws InterruptedExcep
 
   private void givenConnectors(final String... names){
     for (final String name : names) {
-      when(connectClient.describe(name)).thenReturn(ConnectResponse.of(new ConnectorInfo(
+      when(connectClient.describe(name)).thenReturn(ConnectResponse.success(new ConnectorInfo(
           name,
           ImmutableMap.of(),
           ImmutableList.of(),
           ConnectorType.SOURCE
       ), HttpStatus.SC_CREATED));
     }
-    when(connectClient.connectors()).thenReturn(ConnectResponse.of(ImmutableList.copyOf(names),
+    when(connectClient.connectors()).thenReturn(ConnectResponse.success(ImmutableList.copyOf(names),
         HttpStatus.SC_OK));
   }
 

ksql-engine/src/test/java/io/confluent/ksql/services/DefaultConnectClientTest.java

@@ -169,6 +169,23 @@ public void testStatus() throws JsonProcessingException {
     assertThat("Expected no error!", !response.error().isPresent());
   }
 
+  @Test
+  public void testDelete() throws JsonProcessingException {
+    // Given:
+    WireMock.stubFor(
+        WireMock.delete(WireMock.urlEqualTo("/connectors/foo"))
+            .willReturn(WireMock.aResponse()
+                .withStatus(HttpStatus.SC_NO_CONTENT))
+    );
+
+    // When:
+    final ConnectResponse<String> response = client.delete("foo");
+
+    // Then:
+    assertThat(response.datum(), OptionalMatchers.of(is("foo")));
+    assertThat("Expected no error!", !response.error().isPresent());
+  }
+
   @Test
   public void testListShouldRetryOnFailure() throws JsonProcessingException {
     // Given:

ksql-parser/src/main/antlr4/io/confluent/ksql/parser/SqlBase.g4

@@ -63,6 +63,7 @@ statement
     | INSERT INTO qualifiedName (columns)? VALUES values                    #insertValues
     | DROP STREAM (IF EXISTS)? qualifiedName (DELETE TOPIC)?                #dropStream
     | DROP TABLE (IF EXISTS)? qualifiedName  (DELETE TOPIC)?                #dropTable
+    | DROP CONNECTOR identifier                                             #dropConnector
     | EXPLAIN  (statement | qualifiedName)                                  #explain
     | RUN SCRIPT STRING                                                     #runScript
     ;

ksql-parser/src/main/java/io/confluent/ksql/parser/AstBuilder.java

@@ -56,6 +56,7 @@
 import io.confluent.ksql.metastore.model.DataSource;
 import io.confluent.ksql.parser.SqlBaseParser.CreateConnectorContext;
 import io.confluent.ksql.parser.SqlBaseParser.DescribeConnectorContext;
+import io.confluent.ksql.parser.SqlBaseParser.DropConnectorContext;
 import io.confluent.ksql.parser.SqlBaseParser.InsertValuesContext;
 import io.confluent.ksql.parser.SqlBaseParser.IntervalClauseContext;
 import io.confluent.ksql.parser.SqlBaseParser.LimitClauseContext;
@@ -75,6 +76,7 @@
 import io.confluent.ksql.parser.tree.CreateTableAsSelect;
 import io.confluent.ksql.parser.tree.DescribeConnector;
 import io.confluent.ksql.parser.tree.DescribeFunction;
+import io.confluent.ksql.parser.tree.DropConnector;
 import io.confluent.ksql.parser.tree.DropStream;
 import io.confluent.ksql.parser.tree.DropTable;
 import io.confluent.ksql.parser.tree.Explain;
@@ -350,6 +352,14 @@ public Node visitDropStream(final SqlBaseParser.DropStreamContext context) {
       );
     }
 
+    @Override
+    public Node visitDropConnector(final DropConnectorContext context) {
+      return new DropConnector(
+          getLocation(context),
+          ParserUtil.getIdentifierText(context.identifier())
+      );
+    }
+
     @Override
     public Query visitQuery(final SqlBaseParser.QueryContext context) {
       final Relation from = (Relation) visit(context.from);

ksql-parser/src/main/java/io/confluent/ksql/parser/tree/DropConnector.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"; you may not use
+ * this file except in compliance with the License. You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.parser.tree;
+
+import io.confluent.ksql.parser.NodeLocation;
+import java.util.Objects;
+import java.util.Optional;
+
+public class DropConnector extends Statement {
+
+  private final String connectorName;
+
+  public DropConnector(final Optional<NodeLocation> location, final String connectorName) {
+    super(location);
+    this.connectorName = connectorName;
+  }
+
+  public String getConnectorName() {
+    return connectorName;
+  }
+
+  @Override
+  public boolean equals(final Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final DropConnector that = (DropConnector) o;
+    return Objects.equals(connectorName, that.connectorName);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(connectorName);
+  }
+
+  @Override
+  public String toString() {
+    return "DropConnector{"
+        + "connectorName='" + connectorName + '\''
+        + '}';
+  }
+}

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/DropConnectorEntity.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"; you may not use
+ * this file except in compliance with the License. You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.rest.entity;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Objects;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DropConnectorEntity extends KsqlEntity {
+
+  private final String connectorName;
+
+  public DropConnectorEntity(
+      @JsonProperty("statementText") final String statementText,
+      @JsonProperty("connectorName") final String connectorName
+  ) {
+    super(statementText);
+    this.connectorName = Objects.requireNonNull(connectorName, "connectorName");
+  }
+
+  public String getConnectorName() {
+    return connectorName;
+  }
+
+  @Override
+  public String toString() {
+    return "DropConnectorEntity{"
+        + "connectorName='" + connectorName + '\''
+        + '}';
+  }
+}