@Override
    @SuppressWarnings("unchecked")
    public Future<Map<String, Object>> createOrUpdatePutRequest(
            Reconciliation reconciliation,
            String host, int port,
            String connectorName, JsonObject configJson) {
        Buffer data = configJson.toBuffer();
        String path = "/connectors/" + connectorName + "/config";
        LOGGER.debugCr(reconciliation, "Making PUT request to {} with body {}", path, configJson);
        return HttpClientUtils.withHttpClient(vertx, new HttpClientOptions().setLogActivity(true), (httpClient, result) ->
            httpClient.request(HttpMethod.PUT, port, host, path, request -> {
                if (request.succeeded()) {
                    request.result().setFollowRedirects(true)
                            .putHeader("Accept", "application/json")
                            .putHeader("Content-Type", "application/json")
                            .putHeader("Content-Length", String.valueOf(data.length()))
                            .write(data);
                    request.result().send(response -> {
                        if (response.succeeded()) {
                            if (response.result().statusCode() == 200 || response.result().statusCode() == 201) {
                                response.result().bodyHandler(buffer -> {
                                    try {
                                        @SuppressWarnings({ "rawtypes" })
                                        Map t = mapper.readValue(buffer.getBytes(), Map.class);
                                        LOGGER.debugCr(reconciliation, "Got {} response to PUT request to {}: {}", response.result().statusCode(), path, t);
                                        result.complete(t);
                                    } catch (IOException e) {
                                        result.fail(new ConnectRestException(response.result(), "Could not deserialize response: " + e));
                                    }
                                });
                            } else {
                                // TODO Handle 409 (Conflict) indicating a rebalance in progress
                                LOGGER.debugCr(reconciliation, "Got {} response to PUT request to {}", response.result().statusCode(), path);
                                response.result().bodyHandler(buffer -> {
                                    result.fail(new ConnectRestException(response.result(), tryToExtractErrorMessage(reconciliation, buffer)));
                                });
                            }
                        } else {
                            result.tryFail(response.cause());
                        }
                    });
                } else {
                    result.fail(request.cause());
                }
            }));
    }
@Test
    public void testFeatureCompletionWithBadlyFormattedError(Vertx vertx, VertxTestContext context) throws ExecutionException, InterruptedException {
        HttpServer server = mockApi(vertx, 500, "Some error message");

        KafkaConnectApi api = new KafkaConnectApiImpl(vertx);

        Checkpoint async = context.checkpoint();
        api.createOrUpdatePutRequest(Reconciliation.DUMMY_RECONCILIATION, "127.0.0.1", server.actualPort(), "my-connector", new JsonObject())
                        .onComplete(context.failing(res -> context.verify(() -> {
                            assertThat(res.getMessage(), containsString("Unknown error message"));

                            server.close();
                            async.flag();
                        })));
    }