@Override
  public Flux<RawMetric> retrieve(KafkaCluster c, Node node) {
    log.debug("Retrieving metrics from prometheus exporter: {}:{}", node.host(), c.getMetricsConfig().getPort());

    MetricsConfig metricsConfig = c.getMetricsConfig();
    var webClient = new WebClientConfigurator()
        .configureBufferSize(DataSize.ofMegabytes(20))
        .configureBasicAuth(metricsConfig.getUsername(), metricsConfig.getPassword())
        .configureSsl(
            c.getOriginalProperties().getSsl(),
            new ClustersProperties.KeystoreConfig(
                metricsConfig.getKeystoreLocation(),
                metricsConfig.getKeystorePassword()))
        .build();

    return retrieve(webClient, node.host(), c.getMetricsConfig());
  }
@Test
  void callsSecureMetricsEndpointAndConvertsResponceToRawMetric() {
    var url = mockWebServer.url("/metrics");
    mockWebServer.enqueue(prepareResponse());


    MetricsConfig metricsConfig = prepareMetricsConfig(url.port(), "username", "password");

    StepVerifier.create(retriever.retrieve(WebClient.create(), url.host(), metricsConfig))
        .expectNextSequence(expectedRawMetrics())
        // third metric should not be present, since it has "NaN" value
        .verifyComplete();
  }