EOFError when reading response from server

[bdusauso]

• Version of Ruby: 2.4.3 (tested on 2.5.1 as well, same issue) coming from rbenv
• Version of Kafka: 0.11.0.2 (hosted on CloudKarafka. Port number is 9094)
• Version of ruby-kafka: 0.5.5 (tested on 0.6.0beta1, same issue)
• OS version: MacOS 10.13.4

Steps to reproduce

kafka = Kafka.new(
  "ark.srvs.cloudkafka.com:9094",
  client_id: "kafka-example",
  sasl_scram_username: ENV['USERNAME'],
  sasl_scram_password: ENV['PASSWORD'],
  sasl_scram_mechanism: 'sha256'
)
kafka.deliver_message("Hello, World!", topic: "myexample-test")

Expected outcome

The message should be sent to the topic "myexample-test"

Actual outcome

An EOFError has been raised:

Failed to fetch metadata from kafka://ark.srvs.cloudkafka.com:9094: Connection error EOFError: EOFError
/Users/bdusauso/.rbenv/versions/2.4.3/lib/ruby/gems/2.4.0/gems/ruby-kafka-0.5.5/lib/kafka/cluster.rb:371:in `fetch_cluster_info': Could not connect to any of the seed brokers: (Kafka::ConnectionError)
- kafka://ark.srvs.cloudkafka.com:9094: Connection error EOFError: EOFError

Additional informations

Potential source of the problem

It seems the problem comes from an incorrect size being interpreted when the server response is sent back to the client.
The library is trying to read 352518912 bytes from the response, which is seems to be incorrect.

Using rdkafka

I've tried the same tiny example with rdkafkaand it works perfectly.
Here's the snippet:

require 'rdkafka'

config = {
          :"bootstrap.servers" => "ark.srvs.cloudkafka.com:9094" ,
          :"group.id" =>  "kafka-example",
          :"sasl.username" =>  ENV['USERNAME'],
          :"sasl.password"     => ENV['PASSWORD'],
          :"security.protocol" => "SASL_SSL",
          :"sasl.mechanisms"   => "SCRAM-SHA-256"
}

rdkafka = Rdkafka::Config.new(config)
producer = rdkafka.producer
producer.produce(topic: "myexample-test", payload: "Hello World!")

[dasch]

I think you need to enable SSL as well: https://github.com/zendesk/ruby-kafka#encryption-and-authentication-using-ssl. That may be implicitly done in rdkafka, but it has to be explicitly enabled in ruby-kafka.

[snichme]

I tested this out and I think that some wrong assumptions were made.
An SSL socket is only used if any of where params are passed to kafka ca_cert_file_path || ca_cert || client_cert || client_cert_key || ca_certs_from_system but an ssl socket needs to be used for sasl auth as well.
So added a check here: https://github.com/zendesk/ruby-kafka/blob/master/lib/kafka/ssl_context.rb#L7 for any param that identifies a sasl broker will make this work.

[dasch]

@snichme even then, the user needs to pass in a certificate or allow system certs. So it would be better if the SASL code ensured that SSL was enabled, raising an exception otherwise.

[snichme]

Okay, sounds good to me.
Please, just add a note to that on the README, that you have to say which cert to use.