static String getBootstrapServers(Map<String, ?> configs) {
    Object port = configs.get("port");
    String listeners = String.valueOf(configs.get("listeners"));
    if (!"null".equals(listeners) && listeners.length() != 0) {
      // See https://kafka.apache.org/documentation/#listeners for possible responses. If multiple listeners are configured, this function
      // picks the first listener in the list of listeners. Hence, users of this config must adjust their order accordingly.
      String firstListener = listeners.split("\\s*,\\s*")[0];
      String[] protocolHostPort = firstListener.split(":");
      // Use port of listener only if no explicit config specified for KafkaConfig.PortProp().
      String portToUse = port == null ? protocolHostPort[protocolHostPort.length - 1] : String.valueOf(port);
      // Use host of listener if one is specified.
      return ((protocolHostPort[1].length() == 2) ? DEFAULT_BOOTSTRAP_SERVERS_HOST : protocolHostPort[1].substring(2)) + ":" + portToUse;
    }

    return DEFAULT_BOOTSTRAP_SERVERS_HOST + ":" + (port == null ? DEFAULT_BOOTSTRAP_SERVERS_PORT : port);
  }
@Test
  public void testGetKafkaBootstrapServersConfigure() {
    // Test with a "listeners" config with a host
    Map<Object, Object> brokerConfig = buildBrokerConfigs().get(0);
    Map<String, Object> listenersMap = Collections.singletonMap(KafkaConfig.ListenersProp(), brokerConfig.get(KafkaConfig.ListenersProp()));
    String bootstrapServers = CruiseControlMetricsReporter.getBootstrapServers(listenersMap);
    String urlParse = "\\[?([0-9a-zA-Z\\-%._:]*)]?:(-?[0-9]+)";
    Pattern urlParsePattern = Pattern.compile(urlParse);
    assertTrue(urlParsePattern.matcher(bootstrapServers).matches());
    assertEquals(HOST, bootstrapServers.split(":")[0]);

    // Test with a "listeners" config without a host in the first listener.
    String listeners = "SSL://:1234,PLAINTEXT://myhost:4321";
    listenersMap = Collections.singletonMap(KafkaConfig.ListenersProp(), listeners);
    bootstrapServers = CruiseControlMetricsReporter.getBootstrapServers(listenersMap);
    assertTrue(urlParsePattern.matcher(bootstrapServers).matches());
    assertEquals(DEFAULT_BOOTSTRAP_SERVERS_HOST, bootstrapServers.split(":")[0]);
    assertEquals("1234", bootstrapServers.split(":")[1]);

    // Test with "listeners" and "port" config together.
    listenersMap = new HashMap<>();
    listenersMap.put(KafkaConfig.ListenersProp(), listeners);
    listenersMap.put("port", "43");
    bootstrapServers = CruiseControlMetricsReporter.getBootstrapServers(listenersMap);
    assertTrue(urlParsePattern.matcher(bootstrapServers).matches());
    assertEquals(DEFAULT_BOOTSTRAP_SERVERS_HOST, bootstrapServers.split(":")[0]);
    assertEquals("43", bootstrapServers.split(":")[1]);

    // Test with null "listeners" and "port" config.
    bootstrapServers = CruiseControlMetricsReporter.getBootstrapServers(Collections.emptyMap());
    assertTrue(urlParsePattern.matcher(bootstrapServers).matches());
    assertEquals(DEFAULT_BOOTSTRAP_SERVERS_HOST, bootstrapServers.split(":")[0]);
    assertEquals(DEFAULT_BOOTSTRAP_SERVERS_PORT, bootstrapServers.split(":")[1]);
  }