Support SASL OAuthBearer Authentication

[pt2pham]

This PR adds support for OAuthBearer as a method of SASL Authentication as described in KIP-255 and #1710.

When initializing the client, the user must pass in an instance of a class that implements a token method, which returns an ID or Access Token. The interface is described in the documentation for every class where sasl_oauth_token_provider should be configured. I am open to better ideas of where to store the required interface details!

Example:

class TokenProvider(object):
    def token(self):
        return "some_token_id"

producer = KafkaProducer(
    bootstrap_servers='127.0.0.1:9094',
    ...
    security_protocol='SASL_PLAINTEXT',
    sasl_oauth_token_provider=TokenProvider(),
    sasl_mechanism='OAUTHBEARER'
)

Tested with a local Kafka cluster and ID Token, and successfully authenticated with the example client initialization above.

---

This change is 

[dpkp]

Looks great! A few comments in-line for your consideration.

[dpkp]

is string format() going to guarantee correct encoding here? I might suggest operating on and returning bytes instead. E.g., b''.join([b'n,,\x01auth=Bearer ', token, extensions, b'\x01\x01'])

[pt2pham]

I convert to bytes and encode the output of this function in _try_authenticate_oauth so I believe the encoding will be correct.

[dpkp]

interesting -- haven't seen this before

[dpkp]

How does this timeout error work?

[pt2pham]

It's up to the implementer but ideally if the Token Provider pings some OAuth server for a token, if it takes too long it shouldn't go on forever. This was something that I've read/done in other OAuth implementations for different Kafka client libraries.

[dpkp]

Thanks for this!