Minor glitch regarding KSQL configuration

[jeanlouisboudart]

Describe the bug
The documentation (https://github.com/kafka-ops/julie/blob/master/docs/futures/what-ksql-management.rst#configuring-ksql-servers) suggest that we configure the KSQL endpoint as follows :

platform.server.ksql = "http://ksql:8088"

However, this doesn't work and KSQL Client seems to be trying use "http" as a hostname".

To Reproduce
Steps to reproduce the behavior:

1. Setup an environment with Kafka + KSQL
2. Configure platform.server.ksql = "http://ksql:8088" to point to your KSQL instance
3. Create a dummy query
4. Create a dummy topology
5. Run julie ops

Expected behavior
Part of the problem comes from the fact that we parse the provided URL in two places :

1. julie/src/main/java/com/purbon/kafka/topology/JulieOps.java

   Lines 196 to 197 in 73c18ea

   	String server = ksqlAddress.substring(0, ksqlAddress.lastIndexOf(":"));
   	Integer port = Integer.parseInt(ksqlAddress.substring(ksqlAddress.lastIndexOf(":") + 1));

   (at this point the host would be http://ksql and the port would be 8088)

2. https://github.com/kafka-ops/julie/blob/master/src/main/java/com/purbon/kafka/topology/api/ksql/KsqlApiClient.java#L38-L39
   (at this point the host would be http and the port would be 8088)

We should either update the documentation or fix the behavior to be more robust.

Runtime (please complete the following information):

• OS: Ubuntu
• JVM version: openjdk 11
• Version : Master (to test KSQL integration)

[schocco]

To make things more robust, switching to java.net.URL rather then passing Strings around could be an option. The KsqlApiClient constructor could then look like this:

    public KsqlApiClient(URL server, @Nullable BasicAuth basicAuth) {
        this.server = server;
        ClientOptions options = ClientOptions.create()
                .setHost(server.getHost())
                .setPort(server.getPort());
        if (server.getProtocol() != null && server.getProtocol().equals("https")) {
            options.setUseTls(true);
            options.setUseAlpn(true);
        }
        if (basicAuth != null) {
            options.setBasicAuthCredentials(basicAuth.getUser(), basicAuth.getPassword());
        }
        client = Client.create(options);
    }

And the Configuration class would need an additional getter, e.g.

public BasicAuth getKSQLBasicAuth() {
        return new BasicAuth(config.getString(PLATFORM_SERVER_KSQL_BASIC_AUTH_USER), config.getString(PLATFORM_SERVER_KSQL_BASIC_AUTH_PASSWORD));
    }

[purbon]

closed with the merge of #308