KAFKA-4372: Kafka Connect REST API does not handle DELETE of connecto…

…r with slashes in their names

Kafka Connect REST API does not handle in many places connectors with slashes in their names because it expects PathParams, this PR intends to :
* Reject as bad requests API calls trying to create connectors with slashes in their names
* Add support for connector with slashes in their names in the DELETE part of the API to allow users to cleanup their connectors without dropping everything.

This PR adds as well the Unit Test needed for the creation part and was tested manually for the DELETE part.

Author: Olivier Girardot <o.girardot@lateral-thoughts.com>

Reviewers: Shikhar Bhushan <shikhar@confluent.io>, Ewen Cheslack-Postava <ewen@confluent.io>

Closes apache#2096 from ogirardot/fix/connectors-with-slashes-cannot-be-deleted

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java

@@ -89,6 +89,9 @@ public Collection<String> listConnectors(final @QueryParam("forward") Boolean fo
     public Response createConnector(final @QueryParam("forward") Boolean forward,
                                     final CreateConnectorRequest createRequest) throws Throwable {
         String name = createRequest.name();
+        if (name.contains("/")) {
+            throw new BadRequestException("connector name should not contain '/'");
+        }
         Map<String, String> configs = createRequest.config();
         if (!configs.containsKey(ConnectorConfig.NAME_CONFIG))
             configs.put(ConnectorConfig.NAME_CONFIG, name);
@@ -211,7 +214,7 @@ public void restartTask(final @PathParam("connector") String connector,
     }
 
     @DELETE
-    @Path("/{connector}")
+    @Path("/{connector : .+}")
     public void destroyConnector(final @PathParam("connector") String connector,
                                  final @QueryParam("forward") Boolean forward) throws Throwable {
         FutureCallback<Herder.Created<ConnectorInfo>> cb = new FutureCallback<>();

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java

@@ -194,6 +194,23 @@ public void testCreateConnectorExists() throws Throwable {
         PowerMock.verifyAll();
     }
 
+    @Test(expected = BadRequestException.class)
+    public void testCreateConnectorWithASlashInItsName() throws Throwable {
+        String badConnectorName = CONNECTOR_NAME + "/" + "test";
+
+        CreateConnectorRequest body = new CreateConnectorRequest(badConnectorName, Collections.singletonMap(ConnectorConfig.NAME_CONFIG, badConnectorName));
+
+        final Capture<Callback<Herder.Created<ConnectorInfo>>> cb = Capture.newInstance();
+        herder.putConnectorConfig(EasyMock.eq(CONNECTOR_NAME), EasyMock.eq(body.config()), EasyMock.eq(false), EasyMock.capture(cb));
+        expectAndCallbackResult(cb, new Herder.Created<>(true, new ConnectorInfo(CONNECTOR_NAME, CONNECTOR_CONFIG, CONNECTOR_TASK_NAMES)));
+
+        PowerMock.replayAll();
+
+        connectorsResource.createConnector(FORWARD, body);
+
+        PowerMock.verifyAll();
+    }
+
     @Test
     public void testDeleteConnector() throws Throwable {
         final Capture<Callback<Herder.Created<ConnectorInfo>>> cb = Capture.newInstance();