ConnectRestException extends RuntimeException { public int getStatusCode() { return statusCode; } ConnectRestException(String method, String path, int statusCode, String statusMessage, String message);  ConnectRestException(HttpClientResponse response, String message);   ConnectRestException(String method, String path, int statusCode, String statusMessage, String message, Throwable cause);  ConnectRestException(HttpClientResponse response, String message, Throwable cause); int getStatusCode();  }
@Test @SuppressWarnings({"unchecked", "checkstyle:MethodLength", "checkstyle:NPathComplexity"}) public void test(VertxTestContext context) { KafkaConnectApi client = new KafkaConnectApiImpl(vertx); Checkpoint async = context.checkpoint(); client.listConnectorPlugins("localhost", PORT) .onComplete(context.succeeding(connectorPlugins -> context.verify(() -> { assertThat(connectorPlugins.size(), greaterThanOrEqualTo(2)); ConnectorPlugin fileSink = connectorPlugins.stream() .filter(connector -> "org.apache.kafka.connect.file.FileStreamSinkConnector".equals(connector.getConnectorClass())) .findFirst().orElse(null); assertNotNull(fileSink); assertThat(fileSink.getType(), is("sink")); assertThat(fileSink.getVersion(), is(not(emptyString()))); ConnectorPlugin fileSource = connectorPlugins.stream().filter(connector -> "org.apache.kafka.connect.file.FileStreamSourceConnector".equals(connector.getConnectorClass())).findFirst().orElse(null); assertNotNull(fileSource); assertThat(fileSource.getType(), is("source")); assertThat(fileSource.getVersion(), is(not(emptyString()))); }))) .compose(connectorPlugins -> client.list("localhost", PORT)) .onComplete(context.succeeding(connectorNames -> context.verify(() -> assertThat(connectorNames, is(empty()))))) .compose(connectorNames -> { JsonObject o = new JsonObject() .put("connector.class", "FileStreamSource") .put("tasks.max", "1") .put("file", "/dev/null") .put("topic", "my-topic"); return client.createOrUpdatePutRequest("localhost", PORT, "test", o); }) .onComplete(context.succeeding()) .compose(created -> { Promise<Map<String, Object>> promise = Promise.promise(); Handler<Long> handler = new Handler<Long>() { @Override public void handle(Long timerId) { client.status("localhost", PORT, "test").onComplete(result -> { if (result.succeeded()) { Map<String, Object> status = result.result(); if ("RUNNING".equals(((Map) status.getOrDefault("connector", emptyMap())).get("state"))) { promise.complete(status); return; } else { System.err.println(status); } } else { result.cause().printStackTrace(); } vertx.setTimer(1000, this); }); } }; vertx.setTimer(1000, handler); return promise.future(); }) .onComplete(context.succeeding(status -> context.verify(() -> { assertThat(status.get("name"), is("test")); Map<String, Object> connectorStatus = (Map<String, Object>) status.getOrDefault("connector", emptyMap()); assertThat(connectorStatus.get("state"), is("RUNNING")); assertThat(connectorStatus.get("worker_id"), is("localhost:18083")); System.out.println("help " + connectorStatus); List<Map> tasks = (List<Map>) status.get("tasks"); for (Map an : tasks) { assertThat(an.get("state"), is("RUNNING")); assertThat(an.get("worker_id"), is("localhost:18083")); } }))) .compose(status -> client.getConnectorConfig(new BackOff(10), "localhost", PORT, "test")) .onComplete(context.succeeding(config -> context.verify(() -> { assertThat(config, is(TestUtils.map("connector.class", "FileStreamSource", "file", "/dev/null", "tasks.max", "1", "name", "test", "topic", "my-topic"))); }))) .compose(config -> client.getConnectorConfig(new BackOff(10), "localhost", PORT, "does-not-exist")) .onComplete(context.failing(error -> context.verify(() -> { assertThat(error, instanceOf(ConnectRestException.class)); assertThat(((ConnectRestException) error).getStatusCode(), is(404)); }))) .recover(error -> Future.succeededFuture()) .compose(ignored -> client.pause("localhost", PORT, "test")) .onComplete(context.succeeding()) .compose(ignored -> client.resume("localhost", PORT, "test")) .onComplete(context.succeeding()) .compose(ignored -> { JsonObject o = new JsonObject() .put("connector.class", "ThisConnectorDoesNotExist") .put("tasks.max", "1") .put("file", "/dev/null") .put("topic", "my-topic"); return client.createOrUpdatePutRequest("localhost", PORT, "broken", o); }) .onComplete(context.failing(error -> context.verify(() -> { assertThat(error, instanceOf(ConnectRestException.class)); assertThat(error.getMessage(), containsString("Failed to find any class that implements Connector and which name matches ThisConnectorDoesNotExist")); }))) .recover(e -> Future.succeededFuture()) .compose(ignored -> { JsonObject o = new JsonObject() .put("connector.class", "FileStreamSource") .put("tasks.max", "dog") .put("file", "/dev/null") .put("topic", "my-topic"); return client.createOrUpdatePutRequest("localhost", PORT, "broken2", o); }) .onComplete(context.failing(error -> context.verify(() -> { assertThat(error, instanceOf(ConnectRestException.class)); assertThat(error.getMessage(), containsString("Invalid value dog for configuration tasks.max: Not a number of type INT")); }))) .recover(e -> Future.succeededFuture()) .compose(createResponse -> client.list("localhost", PORT)) .onComplete(context.succeeding(connectorNames -> context.verify(() -> assertThat(connectorNames, is(singletonList("test")))))) .compose(connectorNames -> client.delete("localhost", PORT, "test")) .onComplete(context.succeeding()) .compose(deletedConnector -> client.list("localhost", PORT)) .onComplete(context.succeeding(connectorNames -> assertThat(connectorNames, is(empty())))) .compose(connectorNames -> client.delete("localhost", PORT, "never-existed")) .onComplete(context.failing(error -> { assertThat(error, instanceOf(ConnectRestException.class)); assertThat(error.getMessage(), containsString("Connector never-existed not found")); async.flag(); })); }