getBrokerServer() fails to obtain details of Kafka Nodes that use SASL_SSL

[julianharty]

In preliminary testing I discovered there are various assumptions made by the code in getBrokerServer() (in https://github.com/commercetest/pepper-box/blob/05f8a5ddbcfab4b0b2d1992105e3e3a078dc8bb8/src/main/java/com/gslab/pepper/sampler/PepperBoxKafkaSampler.java). These assumptions fail when the Kafka Broker Nodes are configured to use SASL_SSL and the code exits with an unhandled exception.

As a minimum I'd like to handle the exception sufficiently well for the code to use the set of bootstrap servers configured by the user. It might also be worth refining the parsing of the information returned by Zookeeper to use it as that's likely to be more complete and accurately reflect the current state of the cluster than whatever the user of pepper-box has configured.

Note: see related issue #1 for the larger perspective on why fixing the limitation is useful.

[julianharty]

Edited output returned by zk.getData("/brokers/ids/0", ...) etc, when asking for information about a Kafka broker:

String brokerInfo = new String(zk.getData(PropsKeys.BROKER_IDS_ZK_PATH + "/" + id, false, null));

                    /*
                       {"listener_security_protocol_map":{"SASL_SSL":"SASL_SSL"},
                       "endpoints":["SASL_SSL://kafka01:9093"],
                       "jmx_port":-1,
                       "host":null,
                       "timestamp":"1515713063445",
                       "port":-1,
                       "version":4}
                    */

Note: port 9093 is commonly used for the SSL protected port. See also the comments on https://issues.apache.org/jira/browse/KAFKA-3948 which states host and port are only set for backwards compatibility and will not be set for secured endpoints

"The host and port are the for compatibility with old client. The host and port are updated if we have listener with PLAINTEXT. As PLAINTEXT is supported as default."

[julianharty]

And here for comparison is the output from a simple test instance with no security set. Note, the host and port are set in this instance - as relied on by the current code.

/*
  Example output when no security set for Kafka:
  {"jmx_port":-1,
   "timestamp":"1516120590419",
   "endpoints":["PLAINTEXT://127.0.0.1:9092"],
   "host":"127.0.0.1",
   "version":2,
   "port":9092} 
*/

[julianharty]

For completeness here's a similar topic on Stack Overflow which uses the same approach as the existing pepper-box code. https://stackoverflow.com/questions/40683786/how-does-a-kafka-producer-know-which-are-the-available-brokers-from-zookeeper