Add support for SSL, SASL_PLAIN, and SASL_SSL listeners in kafka brokers.

[Crim]

This PR adds support to configure the test brokers to use SSL, SASL_PLAIN, or SASL_SSL authentication methods. By default if no explicit authentication scheme is defined, it will fall back to the previous behavior of using PLAIN.

While the underlying KafkaTestServer supports being configured with multiple authentication schemes/listeners, the exposed Test resources only support a single instance. I believe additional thought needs to be given around how to expose connection details when multiple listener types are available. Likely this would result in breaking changes to the API, so it's been pushed off until the need for this is required.

Example for configuring SSL enabled brokers

When configuring to use SSL, the end user will need to supply paths to their own trust and key stores, as well as passwords. The ones included in these tests are dummy stores created just to validate behavior programatically, and have no use outside of those tests.

    /**
     * We have two node embedded kafka cluster that gets started when this test class is initialized.
     *
     * It's automatically started before any methods are run via the @ClassRule annotation.
     * It's automatically stopped after all of the tests are completed via the @ClassRule annotation.
     *
     * This example we start a cluster with
     *  - 2 brokers (defaults to a single broker)
     *  - configure the brokers to disable topic auto-creation.
     *  - enables SSL authentication with a test/dummy key and trust stores.
     */
    @ClassRule
    public static final SharedKafkaTestResource sharedKafkaTestResource = new SharedKafkaTestResource()
        // Start a cluster with 2 brokers.
        .withBrokers(2)
        // Disable topic auto-creation.
        .withBrokerProperty("auto.create.topics.enable", "false")
        // Register and configure SSL authentication on cluster.
        .registerListener(new SslListener()
        .useSslForInterBrokerProtocol()
        .withKeyStoreLocation(SharedKafkaTestResourceWithSslTest.class.getClassLoader().getResource("kafka.keystore.jks").getFile())
        .withKeyStorePassword("password")
        .withTrustStoreLocation(SharedKafkaTestResourceWithSslTest.class.getClassLoader().getResource("kafka.truststore.jks").getFile())
        .withTrustStorePassword("password")
        .withKeyPassword("password")
    );

Example for configuring SASL_PLAIN enabled brokers

When configuring to use SASL, because of how Kafka reads in JAAS configuration files via an environment variable, the end user will need to start the JVM with the argument: -Djava.security.auth.login.config=path/to/jaas.conf

   /**
     * We have a two node kafka cluster that gets started when this test class is initialized.
     *
     * It's automatically started before any methods are run via the @ClassRule annotation.
     * It's automatically stopped after all of the tests are completed via the @ClassRule annotation.
     *
     * This example we start a cluster with
     *  - 2 brokers (defaults to a single broker)
     *  - configure the brokers to disable topic auto-creation.
     *  - Enable SASL_PLAIN authentication, using username 'kafkaclient' and password 'client-secret'
     */
    @ClassRule
    public static final SharedKafkaTestResource sharedKafkaTestResource = new SharedKafkaTestResource()
        // Start a cluster with 2 brokers.
        .withBrokers(2)
        // Disable topic auto-creation.
        .withBrokerProperty("auto.create.topics.enable", "false")
        // Register and configure SASL PLAIN authentication on cluster.
        .registerListener(new SaslPlainListener()
        .withUsername("kafkaclient")
        .withPassword("client-secret")
    );

Example for configuring SASL_SSL enabled brokers

SASL_SSL is basically a combination of SSL + SASL_PLAIN and also requires the JVM argument -Djava.security.auth.login.config=path/to/jaas.conf

[Crim]

@stanlemon would you mind code reviewing this when you have a moment?

[stanlemon]

This seems like a lot...

[Crim]

hah ya, trying to sort out the OOM errors in travis CI. Odd that it runs without issue locally all the way thru the test suite.

[stanlemon]

Smells like a memory leak somewhere.

[stanlemon]

Can this be final?

[stanlemon]

Disregard, I see now.

[stanlemon]

Is this temporary?

[Crim]

Yea, more trying to figure out the source of the OOM on travis :/

[stanlemon]

Seems pretty straightforward.