@Override
    public ListClientMetricsResourcesResult listClientMetricsResources(ListClientMetricsResourcesOptions options) {
        final long now = time.milliseconds();
        final KafkaFutureImpl<Collection<ClientMetricsResourceListing>> future = new KafkaFutureImpl<>();
        runnable.call(new Call("listClientMetricsResources", calcDeadlineMs(now, options.timeoutMs()),
            new LeastLoadedNodeProvider()) {

            @Override
            ListClientMetricsResourcesRequest.Builder createRequest(int timeoutMs) {
                return new ListClientMetricsResourcesRequest.Builder(new ListClientMetricsResourcesRequestData());
            }

            @Override
            void handleResponse(AbstractResponse abstractResponse) {
                ListClientMetricsResourcesResponse response = (ListClientMetricsResourcesResponse) abstractResponse;
                if (response.error().isFailure()) {
                    future.completeExceptionally(response.error().exception());
                } else {
                    future.complete(response.clientMetricsResources());
                }
            }

            @Override
            void handleFailure(Throwable throwable) {
                future.completeExceptionally(throwable);
            }
        }, now);
        return new ListClientMetricsResourcesResult(future);
    }
@Test
    public void testListClientMetricsResourcesNotSupported() {
        try (AdminClientUnitTestEnv env = mockClientEnv()) {
            env.kafkaClient().prepareResponse(
                request -> request instanceof ListClientMetricsResourcesRequest,
                prepareListClientMetricsResourcesResponse(Errors.UNSUPPORTED_VERSION));

            ListClientMetricsResourcesResult result = env.adminClient().listClientMetricsResources();

            // Validate response
            assertNotNull(result.all());
            TestUtils.assertFutureThrows(result.all(), Errors.UNSUPPORTED_VERSION.exception().getClass());
        }
    }