Support SSL connections

[dpkp]

Adapt #474 to work with new KafkaClient / KafkaConsumer / KafkaProducer. closes #253

[dpkp]

@Ormod do you have some time to test ?

[Ormod]

I'll be able to give it a whirl tomorrow but according to github even this branch has some conflicts.

[dpkp]

Thanks -- I resolved conflicts w/ master (just test cleanup bits), so you should be good to go.

[Ormod]

Tried it out and could make it work at least somewhat. In the good news that with the diff pasted underneath I did get it to at least speak SSL without immediately throwing exceptions.

The reason to use the .values() is that the self._conns is a dict where the key is an integer and the value is the actual connection. The reason for the .pending bit is that since async_connect's maybe_connect() can leave sockets that haven't yet fully connected, they do not get the SSL wrapping to them which would cause them to have a .pending() method.

One thing that hit me badly was that I tried using a utf8 encoded topic name. that didn't really work well. (might want to check that the topic names are actual strings if that's now a requirement)

     # Check for additional pending SSL bytes
     if self.config['security_protocol'] in ('SSL', 'SASL_SSL'):
         # TODO: optimize

•        for conn in self._conns:
  

•            if conn._sock not in ready and conn._sock.pending():
  

•        for conn in self._conns.values():
  

•            if conn._sock not in ready and hasattr(conn._sock, "pending") and conn._sock.pending():
               ready.append(conn._sock)
  

[Ormod]

And oh yeah, as you're doubtless aware, this is conflicting again. Might want to add my diff from above (which github formatted.. in an interesting way) and finally push this to master.

[dpkp]

I've rebased again and done some minimal testing locally. Unfortunately I do not have a reliable SSL cluster for testing. But I'm at least able to get connections and process simple API calls with these latest changes.

Can you elaborate on the utf-8 topic issue? Is that worth opening a new ticket?

[dpkp]

also, apologies that I aggressively rebase pre-commit. I know that makes it harder to pull down changes remotely.

[dpkp]

After a bit more testing, I believe this works and integrates well with the refactored connection management. Merging.