task: metadata: install: pip install -e .[dev]
patch: diff --git a/Makefile b/Makefile
index c0128e7..a624b83 100644
--- a/Makefile
+++ b/Makefile
@@ -2,7 +2,7 @@
 
 SHELL = bash
 
-export KAFKA_VERSION ?= 2.4.0
+export KAFKA_VERSION ?= 4.0.0
 DIST_BASE_URL ?= https://archive.apache.org/dist/kafka/
 
 # Required to support testing old kafka versions on newer java releases
@@ -23,6 +23,9 @@ lint:
 test: build-integration
 	pytest $(PYTESTS)
 
+fixture: build-integration
+	python -m test.fixtures kafka
+
 cov-local: build-integration
 	pytest --pylint --pylint-rcfile=pylint.rc --pylint-error-types=EF --cov=kafka \
 		--cov-config=.covrc --cov-report html $(TEST_FLAGS) kafka test
diff --git a/kafka/client_async.py b/kafka/client_async.py
index 8df4566..8c07110 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -978,8 +978,10 @@ class KafkaClient(object):
                 topics = list(self.config['bootstrap_topics_filter'])
 
             api_version = self.api_version(MetadataRequest, max_version=7)
-            if self.cluster.need_all_topic_metadata or not topics:
+            if self.cluster.need_all_topic_metadata:
                 topics = MetadataRequest[api_version].ALL_TOPICS
+            elif not topics:
+                topics = MetadataRequest[api_version].NO_TOPICS
             if api_version >= 4:
                 request = MetadataRequest[api_version](topics, self.config['allow_auto_create_topics'])
             else:
diff --git a/kafka/protocol/metadata.py b/kafka/protocol/metadata.py
index 3291be8..bb22ba9 100644
--- a/kafka/protocol/metadata.py
+++ b/kafka/protocol/metadata.py
@@ -172,6 +172,7 @@ class MetadataRequest_v0(Request):
         ('topics', Array(String('utf-8')))
     )
     ALL_TOPICS = [] # Empty Array (len 0) for topics returns all topics
+    NO_TOPICS = [] # v0 does not support a 'no topics' request, so we'll just ask for ALL
 
 
 class MetadataRequest_v1(Request):
diff --git a/kafka/sasl/oauth.py b/kafka/sasl/oauth.py
index 4041a93..f1e959c 100644
--- a/kafka/sasl/oauth.py
+++ b/kafka/sasl/oauth.py
@@ -1,10 +1,14 @@
 from __future__ import absolute_import
 
 import abc
+import logging
 
 from kafka.sasl.abc import SaslMechanism
 
 
+log = logging.getLogger(__name__)
+
+
 class SaslMechanismOAuth(SaslMechanism):
 
     def __init__(self, **config):
@@ -12,17 +16,26 @@ class SaslMechanismOAuth(SaslMechanism):
         assert isinstance(config['sasl_oauth_token_provider'], AbstractTokenProvider), \
             'sasl_oauth_token_provider must implement kafka.sasl.oauth.AbstractTokenProvider'
         self.token_provider = config['sasl_oauth_token_provider']
+        self._error = None
         self._is_done = False
         self._is_authenticated = False
 
     def auth_bytes(self):
+        if self._error:
+            # Server should respond to this with SaslAuthenticate failure, which ends the auth process
+            return self._error
         token = self.token_provider.token()
         extensions = self._token_extensions()
         return "n,,\x01auth=Bearer {}{}\x01\x01".format(token, extensions).encode('utf-8')
 
     def receive(self, auth_bytes):
-        self._is_done = True
-        self._is_authenticated = auth_bytes == b''
+        if auth_bytes != b'':
+            error = auth_bytes.decode('utf-8')
+            log.debug("Sending x01 response to server after receiving SASL OAuth error: %s", error)
+            self._error = b'\x01'
+        else:
+            self._is_done = True
+            self._is_authenticated = True
 
     def is_done(self):
         return self._is_done

problem_statement: Producer tries to Describe topics it has no access to
below is my code for producer. My producer is trying to access all the topics and then throwing error 'Denied Operation'. skyDriveProducer has access only to produce messages to qa.dcSacramento.skydrive. But it is trying to access other topics also and throwing error.

Any suggestions?

[2020-06-03 14:25:26,298] INFO Principal = User:skyDriveProducer is Denied Operation = Describe from host = 10.16.19.213 on resource = Topic:LITERAL:__consumer_offsets (kafka.authorizer.logger)
[2020-06-03 14:25:26,298] INFO Principal = User:skyDriveProducer is Denied Operation = Describe from host = 10.16.19.213 on resource = Topic:LITERAL:test (kafka.authorizer.logger)
[2020-06-03 14:25:26,298] INFO Principal = User:skyDriveProducer is Denied Operation = Describe from host = 10.16.19.213 on resource = Topic:LITERAL:qa.dcSacramento.skydrivetest (kafka.authorizer.logger)
[2020-06-03 14:25:26,298] INFO Principal = User:skyDriveProducer is Denied Operation = Describe from host = 10.16.19.213 on resource = Topic:LITERAL:__confluent.support.metrics (kafka.authorizer.logger)

`producer = KafkaProducer(bootstrap_servers=bootstrap_servers,
                                     client_id=client_id,
                                     security_protocol=security_protocol,
                                     sasl_mechanism=sasl_mechanism,
                                     sasl_plain_username=self.username,
                                     sasl_plain_password=self.password,
                                     ssl_check_hostname=False,
                                     acks='all')

            for i in range(int(self.number_of_messages)):
                message = self.message_text + " Current time is " + get_pst_time()
                result = producer.send(self.topic, value=message).get()
                print("sending operation has completed - offset " + str(result.offset) + " topic " + str(result.topic)
                      + " partition " + str(result.partition))

            producer.close()`
repo: dpkp/kafka-python
test_cmd: pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
test_patch: diff --git a/test/test_client_async.py b/test/test_client_async.py
index 8582d8f..2769261 100644
--- a/test/test_client_async.py
+++ b/test/test_client_async.py
@@ -32,7 +32,7 @@ def cli(mocker, conn):
 
 def test_bootstrap(mocker, conn):
     conn.state = ConnectionStates.CONNECTED
-    cli = KafkaClient(api_version=(0, 9))
+    cli = KafkaClient(api_version=(2, 1))
     mocker.patch.object(cli, '_selector')
     future = cli.cluster.request_update()
     cli.poll(future=future)
@@ -43,7 +43,7 @@ def test_bootstrap(mocker, conn):
     kwargs.pop('state_change_callback')
     kwargs.pop('node_id')
     assert kwargs == cli.config
-    conn.send.assert_called_once_with(MetadataRequest[0]([]), blocking=False, request_timeout_ms=None)
+    conn.send.assert_called_once_with(MetadataRequest[7]([], True), blocking=False, request_timeout_ms=None)
     assert cli._bootstrap_fails == 0
     assert cli.cluster.brokers() == set([BrokerMetadata(0, 'foo', 12, None),
                                          BrokerMetadata(1, 'bar', 34, None)])
@@ -330,6 +330,7 @@ def test_maybe_refresh_metadata_update(mocker, client):
     mocker.patch.object(client, 'least_loaded_node', return_value='foobar')
     mocker.patch.object(client, '_can_send_request', return_value=True)
     send = mocker.patch.object(client, 'send')
+    client.cluster.need_all_topic_metadata = True
 
     client.poll(timeout_ms=12345678)
     client._poll.assert_called_with(9999.999) # request_timeout_ms