@VisibleForTesting
  // CHECKSTYLE_RULES.OFF: ParameterNumberCheck
  static ClientOptions createClientOptions(
      // CHECKSTYLE_RULES.ON: ParameterNumberCheck
      final String ksqlServerUrl,
      final String username,
      final String password,
      final String sslTrustStoreLocation,
      final String sslTrustStorePassword,
      final String sslKeystoreLocation,
      final String sslKeystorePassword,
      final String sslKeyPassword,
      final String sslKeyAlias,
      final boolean useAlpn,
      final boolean verifyHost,
      final Map<String, String> requestHeaders
  ) {
    final URL url;
    try {
      url = new URL(ksqlServerUrl);
    } catch (MalformedURLException e) {
      throw new MigrationException("Invalid ksql server URL: " + ksqlServerUrl);
    }

    final ClientOptions options = ClientOptions
        .create()
        .setHost(url.getHost())
        .setPort(url.getPort());

    if (!(username == null || username.isEmpty())
        || !(password == null || password.isEmpty())) {
      options.setBasicAuthCredentials(username, password);
    }

    final boolean useTls = ksqlServerUrl.trim().toLowerCase().startsWith("https://");
    options.setUseTls(useTls);

    if (useTls) {
      options.setTrustStore(sslTrustStoreLocation);
      options.setTrustStorePassword(sslTrustStorePassword);
      options.setKeyStore(sslKeystoreLocation);
      options.setKeyStorePassword(sslKeystorePassword);
      options.setKeyPassword(sslKeyPassword);
      options.setKeyAlias(sslKeyAlias);
      options.setUseAlpn(useAlpn);
      options.setVerifyHost(verifyHost);
    }

    if (requestHeaders != null) {
      options.setRequestHeaders(requestHeaders);
    }

    return options;
  }
@Test
  public void shouldCreateTlsClientOptions() {
    // Given:
    final Map<String, String> requestHeaders = ImmutableMap.of("h1", "v1", "h2", "v2");
    final ClientOptions clientOptions = createClientOptions(TLS_URL, "user",
        "pass", "abc", null, null,
        null, null, null, true, true, requestHeaders);

    // Then:
    assertThat(clientOptions.isUseTls(), is(true));
    assertThat(clientOptions.getBasicAuthUsername(), is("user"));
    assertThat(clientOptions.getBasicAuthPassword(), is("pass"));
    assertThat(clientOptions.getTrustStore(), is("abc"));
    assertThat(clientOptions.getTrustStorePassword(), is(""));
    assertThat(clientOptions.getKeyStore(), is(""));
    assertThat(clientOptions.getKeyStorePassword(), is(""));
    assertThat(clientOptions.getKeyPassword(), is(""));
    assertThat(clientOptions.getKeyAlias(), is(""));
    assertThat(clientOptions.isUseAlpn(), is(true));
    assertThat(clientOptions.isVerifyHost(), is(true));
    assertThat(clientOptions.getRequestHeaders(), is(requestHeaders));
  }