public static URI getServerAddress(final KsqlRestConfig restConfig) {
    final List<String> listeners = restConfig.getList(KsqlRestConfig.LISTENERS_CONFIG);
    final String address = listeners.stream()
        .map(String::trim)
        .findFirst()
        .orElseThrow(() ->
            new ConfigException(KsqlRestConfig.LISTENERS_CONFIG, listeners,
                "value cannot be empty"));

    try {
      return new URL(address).toURI();
    } catch (final Exception e) {
      throw new ConfigException(KsqlRestConfig.LISTENERS_CONFIG, listeners, e.getMessage());
    }
  }
@Test(expected = ConfigException.class)
  public void shouldThrowConfigExceptionIfInvalidServerAddress() {
    // Given:
    final KsqlRestConfig restConfig =
        new KsqlRestConfig(
            Collections.singletonMap(KsqlRestConfig.LISTENERS_CONFIG,
                "invalid"));

    // Then:
    ServerUtil.getServerAddress(restConfig);
  }