@Override
    public Future<Map<String, String>> listConnectLoggers(Reconciliation reconciliation, String host, int port) {
        String path = "/admin/loggers/";
        LOGGER.debugCr(reconciliation, "Making GET request to {}", path);
        return HttpClientUtils.withHttpClient(vertx, new HttpClientOptions().setLogActivity(true), (httpClient, result) ->
                httpClient.request(HttpMethod.GET, port, host, path, request -> {
                    if (request.succeeded()) {
                        request.result().setFollowRedirects(true)
                                .putHeader("Accept", "application/json");
                        request.result().send(response -> {
                            if (response.succeeded()) {
                                if (response.result().statusCode() == 200) {
                                    response.result().bodyHandler(buffer -> {
                                        try {
                                            LOGGER.debugCr(reconciliation, "Got {} response to GET request to {}", response.result().statusCode(), path);
                                            Map<String, Map<String, String>> fetchedLoggers = mapper.readValue(buffer.getBytes(), MAP_OF_MAP_OF_STRINGS);
                                            Map<String, String> loggerMap = new HashMap<>(fetchedLoggers.size());
                                            for (var loggerEntry : fetchedLoggers.entrySet()) {
                                                String level = loggerEntry.getValue().get("level");
                                                if (level != null) {
                                                    loggerMap.put(loggerEntry.getKey(), level);
                                                }
                                            }
                                            result.tryComplete(loggerMap);
                                        } catch (IOException e) {
                                            LOGGER.warnCr(reconciliation, "Failed to get list of connector loggers", e);
                                            result.fail(new ConnectRestException(response.result(), "Failed to get connector loggers", e));
                                        }
                                    });
                                } else {
                                    result.fail(new ConnectRestException(response.result(), "Unexpected status code"));
                                }
                            } else {
                                result.tryFail(response.cause());
                            }
                        });
                    } else {
                        result.tryFail(request.cause());
                    }
                }));
    }
@Test
    public void testListConnectLoggersWithLevelAndLastModified(Vertx vertx, VertxTestContext context) throws Exception {
        final HttpServer server = mockApi(vertx, 200, new ObjectMapper().writeValueAsString(
            Map.of(
                "org.apache.kafka.connect",
                Map.of(
                    "level", "WARN",
                    "last_modified", "2020-01-01T00:00:00.000Z"
                )
            )
        ));
        final KafkaConnectApi api = new KafkaConnectApiImpl(vertx);
        final Checkpoint async = context.checkpoint();
        api.listConnectLoggers(Reconciliation.DUMMY_RECONCILIATION, "127.0.0.1", server.actualPort())
                .onComplete(context.succeeding(res -> context.verify(() -> {
                    assertThat(res, allOf(
                        aMapWithSize(1),
                        hasEntry("org.apache.kafka.connect", "WARN")
                    ));
                    server.close();
                    async.flag();
                })));
    }