Repository: dpkp/kafka-python
Pull request #2478 (2025-02-13 20:52:13)

Always log broker errors in producer.send


Patch:
diff --git a/Makefile b/Makefile
index 3f64d62..0e58387 100644
--- a/Makefile
+++ b/Makefile
@@ -10,6 +10,8 @@ DIST_BASE_URL ?= https://archive.apache.org/dist/kafka/
 # The values here are taken from the 2.4.0 release.
 export KAFKA_JVM_PERFORMANCE_OPTS=-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true -Djava.security.manager=allow
 
+PYTESTS ?= 'test'
+
 setup:
 	pip install -r requirements-dev.txt
 	pip install -Ue .
@@ -18,12 +20,7 @@ lint:
 	pylint --recursive=y --errors-only kafka test
 
 test: build-integration
-	pytest --durations=10 kafka test
-
-# Test using pytest directly if you want to use local python. Useful for other
-# platforms that require manual installation for C libraries, ie. Windows.
-test-local: build-integration
-	pytest --pylint --pylint-rcfile=pylint.rc --pylint-error-types=EF $(TEST_FLAGS) kafka test
+	pytest --durations=10 $(PYTESTS)
 
 cov-local: build-integration
 	pytest --pylint --pylint-rcfile=pylint.rc --pylint-error-types=EF --cov=kafka \
@@ -92,7 +89,7 @@ servers/dist/jakarta.xml.bind-api-2.3.3.jar:
 servers/%/kafka-bin: servers/dist/$$(call kafka_artifact_name,$$*) | servers/dist
 	@echo "Extracting kafka $* binaries from $<"
 	if [ -d "$@" ]; then rm -rf $@.bak; mv $@ $@.bak; fi
-	mkdir $@
+	mkdir -p $@
 	tar xzvf $< -C $@ --strip-components 1
 	if [[ "$*" < "1" ]]; then make servers/patch-libs/$*; fi
 
diff --git a/kafka/admin/client.py b/kafka/admin/client.py
index 22c2987..6252783 100644
--- a/kafka/admin/client.py
+++ b/kafka/admin/client.py
@@ -72,7 +72,7 @@ class KafkaAdminClient(object):
             reconnection attempts will continue periodically with this fixed
             rate. To avoid connection storms, a randomization factor of 0.2
             will be applied to the backoff resulting in a random range between
-            20% below and 20% above the computed value. Default: 1000.
+            20% below and 20% above the computed value. Default: 30000.
         request_timeout_ms (int): Client request timeout in milliseconds.
             Default: 30000.
         connections_max_idle_ms: Close idle connections after the number of
@@ -156,7 +156,7 @@ class KafkaAdminClient(object):
         'request_timeout_ms': 30000,
         'connections_max_idle_ms': 9 * 60 * 1000,
         'reconnect_backoff_ms': 50,
-        'reconnect_backoff_max_ms': 1000,
+        'reconnect_backoff_max_ms': 30000,
         'max_in_flight_requests_per_connection': 5,
         'receive_buffer_bytes': None,
         'send_buffer_bytes': None,
diff --git a/kafka/client_async.py b/kafka/client_async.py
index 58f22d4..1bde074 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -75,7 +75,7 @@ class KafkaClient(object):
             reconnection attempts will continue periodically with this fixed
             rate. To avoid connection storms, a randomization factor of 0.2
             will be applied to the backoff resulting in a random range between
-            20% below and 20% above the computed value. Default: 1000.
+            20% below and 20% above the computed value. Default: 30000.
         request_timeout_ms (int): Client request timeout in milliseconds.
             Default: 30000.
         connections_max_idle_ms: Close idle connections after the number of
@@ -164,7 +164,7 @@ class KafkaClient(object):
         'wakeup_timeout_ms': 3000,
         'connections_max_idle_ms': 9 * 60 * 1000,
         'reconnect_backoff_ms': 50,
-        'reconnect_backoff_max_ms': 1000,
+        'reconnect_backoff_max_ms': 30000,
         'max_in_flight_requests_per_connection': 5,
         'receive_buffer_bytes': None,
         'send_buffer_bytes': None,
@@ -204,8 +204,9 @@ class KafkaClient(object):
         # these properties need to be set on top of the initialization pipeline
         # because they are used when __del__ method is called
         self._closed = False
-        self._wake_r, self._wake_w = socket.socketpair()
         self._selector = self.config['selector']()
+        self._init_wakeup_socketpair()
+        self._wake_lock = threading.Lock()
 
         self.cluster = ClusterMetadata(**self.config)
         self._topics = set()  # empty set will fetch all topic metadata
@@ -217,9 +218,6 @@ class KafkaClient(object):
         self._refresh_on_disconnects = True
         self._last_bootstrap = 0
         self._bootstrap_fails = 0
-        self._wake_r.setblocking(False)
-        self._wake_w.settimeout(self.config['wakeup_timeout_ms'] / 1000.0)
-        self._wake_lock = threading.Lock()
 
         self._lock = threading.RLock()
 
@@ -228,7 +226,6 @@ class KafkaClient(object):
         # lock above.
         self._pending_completion = collections.deque()
 
-        self._selector.register(self._wake_r, selectors.EVENT_READ)
         self._idle_expiry_manager = IdleConnectionManager(self.config['connections_max_idle_ms'])
         self._sensors = None
         if self.config['metrics']:
@@ -243,6 +240,25 @@ class KafkaClient(object):
             check_timeout = self.config['api_version_auto_timeout_ms'] / 1000
             self.config['api_version'] = self.check_version(timeout=check_timeout)
 
+    def _init_wakeup_socketpair(self):
+        self._wake_r, self._wake_w = socket.socketpair()
+        self._wake_r.setblocking(False)
+        self._wake_w.settimeout(self.config['wakeup_timeout_ms'] / 1000.0)
+        self._waking = False
+        self._selector.register(self._wake_r, selectors.EVENT_READ)
+
+    def _close_wakeup_socketpair(self):
+        if self._wake_r is not None:
+            try:
+                self._selector.unregister(self._wake_r)
+            except (KeyError, ValueError, TypeError):
+                pass
+            self._wake_r.close()
+        if self._wake_w is not None:
+            self._wake_w.close()
+        self._wake_r = None
+        self._wake_w = None
+
     def _can_bootstrap(self):
         effective_failures = self._bootstrap_fails // self._num_bootstrap_hosts
         backoff_factor = 2 ** effective_failures
@@ -416,8 +432,7 @@ class KafkaClient(object):
     def _close(self):
         if not self._closed:
             self._closed = True
-            self._wake_r.close()
-            self._wake_w.close()
+            self._close_wakeup_socketpair()
             self._selector.close()
 
     def close(self, node_id=None):
@@ -464,9 +479,8 @@ class KafkaClient(object):
     def connection_delay(self, node_id):
         """
         Return the number of milliseconds to wait, based on the connection
-        state, before attempting to send data. When disconnected, this respects
-        the reconnect backoff time. When connecting, returns 0 to allow
-        non-blocking connect to finish. When connected, returns a very large
+        state, before attempting to send data. When connecting or disconnected,
+        this respects the reconnect backoff time. When connected, returns a very large
         number to handle slow/stalled connections.
 
         Arguments:
@@ -537,7 +551,8 @@ class KafkaClient(object):
         # we will need to call send_pending_requests()
         # to trigger network I/O
         future = conn.send(request, blocking=False)
-        self._sending.add(conn)
+        if not future.is_done:
+            self._sending.add(conn)
 
         # Wakeup signal is useful in case another thread is
         # blocked waiting for incoming network traffic while holding
@@ -563,9 +578,7 @@ class KafkaClient(object):
         Returns:
             list: responses received (can be empty)
         """
-        if future is not None:
-            timeout_ms = 100
-        elif timeout_ms is None:
+        if timeout_ms is None:
             timeout_ms = self.config['request_timeout_ms']
         elif not isinstance(timeout_ms, (int, float)):
             raise TypeError('Invalid type for timeout: %s' % type(timeout_ms))
@@ -577,26 +590,25 @@ class KafkaClient(object):
                 if self._closed:
                     break
 
+                # Send a metadata request if needed (or initiate new connection)
+                metadata_timeout_ms = self._maybe_refresh_metadata()
+
                 # Attempt to complete pending connections
                 for node_id in list(self._connecting):
                     self._maybe_connect(node_id)
 
-                # Send a metadata request if needed
-                metadata_timeout_ms = self._maybe_refresh_metadata()
-
                 # If we got a future that is already done, don't block in _poll
                 if future is not None and future.is_done:
                     timeout = 0
                 else:
                     idle_connection_timeout_ms = self._idle_expiry_manager.next_check_ms()
+                    request_timeout_ms = self._next_ifr_request_timeout_ms()
+                    log.debug("Timeouts: user %f, metadata %f, idle connection %f, request %f", timeout_ms, metadata_timeout_ms, idle_connection_timeout_ms, request_timeout_ms)
                     timeout = min(
                         timeout_ms,
                         metadata_timeout_ms,
                         idle_connection_timeout_ms,
-                        self.config['request_timeout_ms'])
-                    # if there are no requests in flight, do not block longer than the retry backoff
-                    if self.in_flight_request_count() == 0:
-                        timeout = min(timeout, self.config['retry_backoff_ms'])
+                        request_timeout_ms)
                     timeout = max(0, timeout)  # avoid negative timeouts
 
                 self._poll(timeout / 1000)
@@ -615,6 +627,8 @@ class KafkaClient(object):
     def _register_send_sockets(self):
         while self._sending:
             conn = self._sending.pop()
+            if conn._sock is None:
+                continue
             try:
                 key = self._selector.get_key(conn._sock)
                 events = key.events | selectors.EVENT_WRITE
@@ -772,6 +786,17 @@ class KafkaClient(object):
 
         return found
 
+    def least_loaded_node_refresh_ms(self):
+        """Return connection delay in milliseconds for next available node.
+
+        This method is used primarily for retry/backoff during metadata refresh
+        during / after a cluster outage, in which there are no available nodes.
+
+        Returns:
+           float: delay_ms
+        """
+        return min([self.connection_delay(broker.nodeId) for broker in self.cluster.brokers()])
+
     def set_topics(self, topics):
         """Set specific topics to track for metadata.
 
@@ -803,12 +828,18 @@ class KafkaClient(object):
         self._topics.add(topic)
         return self.cluster.request_update()
 
+    def _next_ifr_request_timeout_ms(self):
+        if self._conns:
+            return min([conn.next_ifr_request_timeout_ms() for conn in six.itervalues(self._conns)])
+        else:
+            return float('inf')
+
     # This method should be locked when running multi-threaded
     def _maybe_refresh_metadata(self, wakeup=False):
         """Send a metadata request if needed.
 
         Returns:
-            int: milliseconds until next refresh
+            float: milliseconds until next refresh
         """
         ttl = self.cluster.ttl()
         wait_for_in_progress_ms = self.config['request_timeout_ms'] if self._metadata_refresh_in_progress else 0
@@ -822,8 +853,9 @@ class KafkaClient(object):
         # least_loaded_node()
         node_id = self.least_loaded_node()
         if node_id is None:
-            log.debug("Give up sending metadata request since no node is available");
-            return self.config['reconnect_backoff_ms']
+            next_connect_ms = self.least_loaded_node_refresh_ms()
+            log.debug("Give up sending metadata request since no node is available. (reconnect delay %d ms)", next_connect_ms)
+            return next_connect_ms
 
         if self._can_send_request(node_id):
             topics = list(self._topics)
@@ -850,11 +882,11 @@ class KafkaClient(object):
         # the client from unnecessarily connecting to additional nodes while a previous connection
         # attempt has not been completed.
         if self._connecting:
-            return self.config['reconnect_backoff_ms']
+            return float('inf')
 
         if self.maybe_connect(node_id, wakeup=wakeup):
             log.debug("Initializing connection to node %s for metadata request", node_id)
-            return self.config['reconnect_backoff_ms']
+            return float('inf')
 
         # connected but can't send more, OR connecting
         # In either case we just need to wait for a network event
@@ -927,22 +959,34 @@ class KafkaClient(object):
             raise Errors.NoBrokersAvailable()
 
     def wakeup(self):
+        if self._waking or self._wake_w is None:
+            return
         with self._wake_lock:
             try:
                 self._wake_w.sendall(b'x')
-            except socket.timeout:
+                self._waking = True
+            except socket.timeout as e:
                 log.warning('Timeout to send to wakeup socket!')
-                raise Errors.KafkaTimeoutError()
-            except socket.error:
-                log.warning('Unable to send to wakeup socket!')
+                raise Errors.KafkaTimeoutError(e)
+            except socket.error as e:
+                log.warning('Unable to send to wakeup socket! %s', e)
+                raise e
 
     def _clear_wake_fd(self):
         # reading from wake socket should only happen in a single thread
-        while True:
-            try:
-                self._wake_r.recv(1024)
-            except socket.error:
-                break
+        with self._wake_lock:
+            self._waking = False
+            while True:
+                try:
+                    if not self._wake_r.recv(1024):
+                        # Non-blocking socket returns empty on error
+                        log.warning("Error reading wakeup socket. Rebuilding socketpair.")
+                        self._close_wakeup_socketpair()
+                        self._init_wakeup_socketpair()
+                        break
+                except socket.error:
+                    # Non-blocking socket raises when socket is ok but no data available to read
+                    break
 
     def _maybe_close_oldest_connection(self):
         expired_connection = self._idle_expiry_manager.poll_expired_connection()
diff --git a/kafka/cluster.py b/kafka/cluster.py
index 438baf2..4b07cc7 100644
--- a/kafka/cluster.py
+++ b/kafka/cluster.py
@@ -236,7 +236,7 @@ class ClusterMetadata(object):
         """
         # In the common case where we ask for a single topic and get back an
         # error, we should fail the future
-        if len(metadata.topics) == 1 and metadata.topics[0][0] != 0:
+        if len(metadata.topics) == 1 and metadata.topics[0][0] != Errors.NoError.errno:
             error_code, topic = metadata.topics[0][:2]
             error = Errors.for_code(error_code)(topic)
             return self.failed_update(error)
diff --git a/kafka/codec.py b/kafka/codec.py
index c740a18..b73df06 100644
--- a/kafka/codec.py
+++ b/kafka/codec.py
@@ -193,8 +193,15 @@ def _detect_xerial_stream(payload):
     """
 
     if len(payload) > 16:
-        header = struct.unpack('!' + _XERIAL_V1_FORMAT, bytes(payload)[:16])
-        return header == _XERIAL_V1_HEADER
+        magic = struct.unpack('!' + _XERIAL_V1_FORMAT[:8], bytes(payload)[:8])
+        version, compat = struct.unpack('!' + _XERIAL_V1_FORMAT[8:], bytes(payload)[8:16])
+        # Until there is more than one way to do xerial blocking, the version + compat
+        # fields can be ignored. Also some producers (i.e., redpanda) are known to
+        # incorrectly encode these as little-endian, and that causes us to fail decoding
+        # when we otherwise would have succeeded.
+        # See https://github.com/dpkp/kafka-python/issues/2414
+        if magic == _XERIAL_V1_HEADER[:8]:
+            return True
     return False
 
 
diff --git a/kafka/conn.py b/kafka/conn.py
index 4a43976..7dab799 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -120,7 +120,7 @@ class BrokerConnection(object):
             reconnection attempts will continue periodically with this fixed
             rate. To avoid connection storms, a randomization factor of 0.2
             will be applied to the backoff resulting in a random range between
-            20% below and 20% above the computed value. Default: 1000.
+            20% below and 20% above the computed value. Default: 30000.
         request_timeout_ms (int): Client request timeout in milliseconds.
             Default: 30000.
         max_in_flight_requests_per_connection (int): Requests are pipelined
@@ -198,7 +198,7 @@ class BrokerConnection(object):
         'node_id': 0,
         'request_timeout_ms': 30000,
         'reconnect_backoff_ms': 50,
-        'reconnect_backoff_max_ms': 1000,
+        'reconnect_backoff_max_ms': 30000,
         'max_in_flight_requests_per_connection': 5,
         'receive_buffer_bytes': None,
         'send_buffer_bytes': None,
@@ -848,20 +848,22 @@ class BrokerConnection(object):
         re-establish a connection yet
         """
         if self.state is ConnectionStates.DISCONNECTED:
-            if time.time() < self.last_attempt + self._reconnect_backoff:
-                return True
+            return self.connection_delay() > 0
         return False
 
     def connection_delay(self):
         """
         Return the number of milliseconds to wait, based on the connection
-        state, before attempting to send data. When disconnected, this respects
-        the reconnect backoff time. When connecting or connected, returns a very
+        state, before attempting to send data. When connecting or disconnected,
+        this respects the reconnect backoff time. When connected, returns a very
         large number to handle slow/stalled connections.
         """
-        time_waited = time.time() - (self.last_attempt or 0)
-        if self.state is ConnectionStates.DISCONNECTED:
-            return max(self._reconnect_backoff - time_waited, 0) * 1000
+        if self.disconnected() or self.connecting():
+            if len(self._gai) > 0:
+                return 0
+            else:
+                time_waited = time.time() - self.last_attempt
+                return max(self._reconnect_backoff - time_waited, 0) * 1000
         else:
             # When connecting or connected, we should be able to delay
             # indefinitely since other events (connection or data acked) will
@@ -887,6 +889,9 @@ class BrokerConnection(object):
         self._failures = 0
         self._reconnect_backoff = self.config['reconnect_backoff_ms'] / 1000.0
 
+    def _reconnect_jitter_pct(self):
+        return uniform(0.8, 1.2)
+
     def _update_reconnect_backoff(self):
         # Do not mark as failure if there are more dns entries available to try
         if len(self._gai) > 0:
@@ -895,7 +900,7 @@ class BrokerConnection(object):
             self._failures += 1
             self._reconnect_backoff = self.config['reconnect_backoff_ms'] * 2 ** (self._failures - 1)
             self._reconnect_backoff = min(self._reconnect_backoff, self.config['reconnect_backoff_max_ms'])
-            self._reconnect_backoff *= uniform(0.8, 1.2)
+            self._reconnect_backoff *= self._reconnect_jitter_pct()
             self._reconnect_backoff /= 1000.0
             log.debug('%s: reconnect backoff %s after %s failures', self, self._reconnect_backoff, self._failures)
 
@@ -1136,15 +1141,18 @@ class BrokerConnection(object):
         return ()
 
     def requests_timed_out(self):
+        return self.next_ifr_request_timeout_ms() == 0
+
+    def next_ifr_request_timeout_ms(self):
         with self._lock:
             if self.in_flight_requests:
                 get_timestamp = lambda v: v[1]
                 oldest_at = min(map(get_timestamp,
                                     self.in_flight_requests.values()))
-                timeout = self.config['request_timeout_ms'] / 1000.0
-                if time.time() >= oldest_at + timeout:
-                    return True
-            return False
+                next_timeout = oldest_at + self.config['request_timeout_ms'] / 1000.0
+                return max(0, (next_timeout - time.time()) * 1000)
+            else:
+                return float('inf')
 
     def _handle_api_version_response(self, response):
         error_type = Errors.for_code(response.error_code)
@@ -1255,13 +1263,14 @@ class BrokerConnection(object):
             # request was unrecognized
             mr = self.send(MetadataRequest[0](topics))
 
-            selector = self.config['selector']()
-            selector.register(self._sock, selectors.EVENT_READ)
-            while not (f.is_done and mr.is_done):
-                selector.select(1)
-                for response, future in self.recv():
-                    future.success(response)
-            selector.close()
+            if not (f.is_done and mr.is_done) and self._sock is not None:
+                selector = self.config['selector']()
+                selector.register(self._sock, selectors.EVENT_READ)
+                while not (f.is_done and mr.is_done):
+                    selector.select(1)
+                    for response, future in self.recv():
+                        future.success(response)
+                selector.close()
 
             if f.succeeded():
                 if isinstance(request, ApiVersionRequest[0]):
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index a1d1dfa..2d7571d 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -98,7 +98,7 @@ class KafkaConsumer(six.Iterator):
             reconnection attempts will continue periodically with this fixed
             rate. To avoid connection storms, a randomization factor of 0.2
             will be applied to the backoff resulting in a random range between
-            20% below and 20% above the computed value. Default: 1000.
+            20% below and 20% above the computed value. Default: 30000.
         max_in_flight_requests_per_connection (int): Requests are pipelined
             to kafka brokers up to this number of maximum requests per
             broker connection. Default: 5.
@@ -263,7 +263,7 @@ class KafkaConsumer(six.Iterator):
         'request_timeout_ms': 305000, # chosen to be higher than the default of max_poll_interval_ms
         'retry_backoff_ms': 100,
         'reconnect_backoff_ms': 50,
-        'reconnect_backoff_max_ms': 1000,
+        'reconnect_backoff_max_ms': 30000,
         'max_in_flight_requests_per_connection': 5,
         'auto_offset_reset': 'latest',
         'enable_auto_commit': True,
diff --git a/kafka/errors.py b/kafka/errors.py
index b33cf51..b8fa067 100644
--- a/kafka/errors.py
+++ b/kafka/errors.py
@@ -186,7 +186,8 @@ class ReplicaNotAvailableError(BrokerResponseError):
     message = 'REPLICA_NOT_AVAILABLE'
     description = ('If replica is expected on a broker, but is not (this can be'
                    ' safely ignored).')
-
+    retriable = True
+    invalid_metadata = True
 
 class MessageSizeTooLargeError(BrokerResponseError):
     errno = 10
@@ -210,10 +211,11 @@ class OffsetMetadataTooLargeError(BrokerResponseError):
                    ' offset metadata.')
 
 
-# TODO is this deprecated? https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes
-class StaleLeaderEpochCodeError(BrokerResponseError):
+class NetworkExceptionError(BrokerResponseError):
     errno = 13
-    message = 'STALE_LEADER_EPOCH_CODE'
+    message = 'NETWORK_EXCEPTION'
+    retriable = True
+    invalid_metadata = True
 
 
 class GroupLoadInProgressError(BrokerResponseError):
@@ -441,24 +443,596 @@ class PolicyViolationError(BrokerResponseError):
     errno = 44
     message = 'POLICY_VIOLATION'
     description = 'Request parameters do not satisfy the configured policy.'
+    retriable = False
+
+
+class OutOfOrderSequenceNumberError(BrokerResponseError):
+    errno = 45
+    message = 'OUT_OF_ORDER_SEQUENCE_NUMBER'
+    description = 'The broker received an out of order sequence number.'
+    retriable = False
+
+
+class DuplicateSequenceNumberError(BrokerResponseError):
+    errno = 46
+    message = 'DUPLICATE_SEQUENCE_NUMBER'
+    description = 'The broker received a duplicate sequence number.'
+    retriable = False
+
+
+class InvalidProducerEpochError(BrokerResponseError):
+    errno = 47
+    message = 'INVALID_PRODUCER_EPOCH'
+    description = 'Producer attempted to produce with an old epoch.'
+    retriable = False
+
+
+class InvalidTxnStateError(BrokerResponseError):
+    errno = 48
+    message = 'INVALID_TXN_STATE'
+    description = 'The producer attempted a transactional operation in an invalid state.'
+    retriable = False
+
+
+class InvalidProducerIdMappingError(BrokerResponseError):
+    errno = 49
+    message = 'INVALID_PRODUCER_ID_MAPPING'
+    description = 'The producer attempted to use a producer id which is not currently assigned to its transactional id.'
+    retriable = False
+
+
+class InvalidTransactionTimeoutError(BrokerResponseError):
+    errno = 50
+    message = 'INVALID_TRANSACTION_TIMEOUT'
+    description = 'The transaction timeout is larger than the maximum value allowed by the broker (as configured by transaction.max.timeout.ms).'
+    retriable = False
+
+
+class ConcurrentTransactionsError(BrokerResponseError):
+    errno = 51
+    message = 'CONCURRENT_TRANSACTIONS'
+    description = 'The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing.'
+    retriable = True
+
+
+class TransactionCoordinatorFencedError(BrokerResponseError):
+    errno = 52
+    message = 'TRANSACTION_COORDINATOR_FENCED'
+    description = 'Indicates that the transaction coordinator sending a WriteTxnMarker is no longer the current coordinator for a given producer.'
+    retriable = False
+
+
+class TransactionalIdAuthorizationFailedError(BrokerResponseError):
+    errno = 53
+    message = 'TRANSACTIONAL_ID_AUTHORIZATION_FAILED'
+    description = 'Transactional Id authorization failed.'
+    retriable = False
 
 
 class SecurityDisabledError(BrokerResponseError):
     errno = 54
     message = 'SECURITY_DISABLED'
     description = 'Security features are disabled.'
+    retriable = False
+
+
+class OperationNotAttemptedError(BrokerResponseError):
+    errno = 55
+    message = 'OPERATION_NOT_ATTEMPTED'
+    description = 'The broker did not attempt to execute this operation. This may happen for batched RPCs where some operations in the batch failed, causing the broker to respond without trying the rest.'
+    retriable = False
+
+
+class KafkaStorageError(BrokerResponseError):
+    errno = 56
+    message = 'KAFKA_STORAGE_ERROR'
+    description = 'Disk error when trying to access log file on the disk.'
+    retriable = True
+    invalid_metadata = True
+
+
+class LogDirNotFoundError(BrokerResponseError):
+    errno = 57
+    message = 'LOG_DIR_NOT_FOUND'
+    description = 'The user-specified log directory is not found in the broker config.'
+    retriable = False
+
+
+class SaslAuthenticationFailedError(BrokerResponseError):
+    errno = 58
+    message = 'SASL_AUTHENTICATION_FAILED'
+    description = 'SASL Authentication failed.'
+    retriable = False
+
+
+class UnknownProducerIdError(BrokerResponseError):
+    errno = 59
+    message = 'UNKNOWN_PRODUCER_ID'
+    description = 'This exception is raised by the broker if it could not locate the producer metadata associated with the producerId in question. This could happen if, for instance, the producer\'s records were deleted because their retention time had elapsed. Once the last records of the producerId are removed, the producer\'s metadata is removed from the broker, and future appends by the producer will return this exception.'
+    retriable = False
+
+
+class ReassignmentInProgressError(BrokerResponseError):
+    errno = 60
+    message = 'REASSIGNMENT_IN_PROGRESS'
+    description = 'A partition reassignment is in progress.'
+    retriable = False
+
+
+class DelegationTokenAuthDisabledError(BrokerResponseError):
+    errno = 61
+    message = 'DELEGATION_TOKEN_AUTH_DISABLED'
+    description = 'Delegation Token feature is not enabled.'
+    retriable = False
+
+
+class DelegationTokenNotFoundError(BrokerResponseError):
+    errno = 62
+    message = 'DELEGATION_TOKEN_NOT_FOUND'
+    description = 'Delegation Token is not found on server.'
+    retriable = False
+
+
+class DelegationTokenOwnerMismatchError(BrokerResponseError):
+    errno = 63
+    message = 'DELEGATION_TOKEN_OWNER_MISMATCH'
+    description = 'Specified Principal is not valid Owner/Renewer.'
+    retriable = False
+
+
+class DelegationTokenRequestNotAllowedError(BrokerResponseError):
+    errno = 64
+    message = 'DELEGATION_TOKEN_REQUEST_NOT_ALLOWED'
+    description = 'Delegation Token requests are not allowed on PLAINTEXT/1-way SSL channels and on delegation token authenticated channels.'
+    retriable = False
+
+
+class DelegationTokenAuthorizationFailedError(BrokerResponseError):
+    errno = 65
+    message = 'DELEGATION_TOKEN_AUTHORIZATION_FAILED'
+    description = 'Delegation Token authorization failed.'
+    retriable = False
+
+
+class DelegationTokenExpiredError(BrokerResponseError):
+    errno = 66
+    message = 'DELEGATION_TOKEN_EXPIRED'
+    description = 'Delegation Token is expired.'
+    retriable = False
+
+
+class InvalidPrincipalTypeError(BrokerResponseError):
+    errno = 67
+    message = 'INVALID_PRINCIPAL_TYPE'
+    description = 'Supplied principalType is not supported.'
+    retriable = False
 
 
 class NonEmptyGroupError(BrokerResponseError):
     errno = 68
     message = 'NON_EMPTY_GROUP'
     description = 'The group is not empty.'
+    retriable = False
 
 
 class GroupIdNotFoundError(BrokerResponseError):
     errno = 69
     message = 'GROUP_ID_NOT_FOUND'
     description = 'The group id does not exist.'
+    retriable = False
+
+
+class FetchSessionIdNotFoundError(BrokerResponseError):
+    errno = 70
+    message = 'FETCH_SESSION_ID_NOT_FOUND'
+    description = 'The fetch session ID was not found.'
+    retriable = True
+
+
+class InvalidFetchSessionEpochError(BrokerResponseError):
+    errno = 71
+    message = 'INVALID_FETCH_SESSION_EPOCH'
+    description = 'The fetch session epoch is invalid.'
+    retriable = True
+
+
+class ListenerNotFoundError(BrokerResponseError):
+    errno = 72
+    message = 'LISTENER_NOT_FOUND'
+    description = 'There is no listener on the leader broker that matches the listener on which metadata request was processed.'
+    retriable = True
+    invalid_metadata = True
+
+
+class TopicDeletionDisabledError(BrokerResponseError):
+    errno = 73
+    message = 'TOPIC_DELETION_DISABLED'
+    description = 'Topic deletion is disabled.'
+    retriable = False
+
+
+class FencedLeaderEpochError(BrokerResponseError):
+    errno = 74
+    message = 'FENCED_LEADER_EPOCH'
+    description = 'The leader epoch in the request is older than the epoch on the broker.'
+    retriable = True
+    invalid_metadata = True
+
+
+class UnknownLeaderEpochError(BrokerResponseError):
+    errno = 75
+    message = 'UNKNOWN_LEADER_EPOCH'
+    description = 'The leader epoch in the request is newer than the epoch on the broker.'
+    retriable = True
+
+
+class UnsupportedCompressionTypeError(BrokerResponseError):
+    errno = 76
+    message = 'UNSUPPORTED_COMPRESSION_TYPE'
+    description = 'The requesting client does not support the compression type of given partition.'
+    retriable = False
+
+
+class StaleBrokerEpochError(BrokerResponseError):
+    errno = 77
+    message = 'STALE_BROKER_EPOCH'
+    description = 'Broker epoch has changed.'
+    retriable = False
+
+
+class OffsetNotAvailableError(BrokerResponseError):
+    errno = 78
+    message = 'OFFSET_NOT_AVAILABLE'
+    description = 'The leader high watermark has not caught up from a recent leader election so the offsets cannot be guaranteed to be monotonically increasing.'
+    retriable = True
+
+
+class MemberIdRequiredError(BrokerResponseError):
+    errno = 79
+    message = 'MEMBER_ID_REQUIRED'
+    description = 'The group member needs to have a valid member id before actually entering a consumer group.'
+    retriable = False
+
+
+class PreferredLeaderNotAvailableError(BrokerResponseError):
+    errno = 80
+    message = 'PREFERRED_LEADER_NOT_AVAILABLE'
+    description = 'The preferred leader was not available.'
+    retriable = True
+    invalid_metadata = True
+
+
+class GroupMaxSizeReachedError(BrokerResponseError):
+    errno = 81
+    message = 'GROUP_MAX_SIZE_REACHED'
+    description = 'The consumer group has reached its max size.'
+    retriable = False
+
+
+class FencedInstanceIdError(BrokerResponseError):
+    errno = 82
+    message = 'FENCED_INSTANCE_ID'
+    description = 'The broker rejected this static consumer since another consumer with the same group.instance.id has registered with a different member.id.'
+    retriable = False
+
+
+class EligibleLeadersNotAvailableError(BrokerResponseError):
+    errno = 83
+    message = 'ELIGIBLE_LEADERS_NOT_AVAILABLE'
+    description = 'Eligible topic partition leaders are not available.'
+    retriable = True
+    invalid_metadata = True
+
+
+class ElectionNotNeededError(BrokerResponseError):
+    errno = 84
+    message = 'ELECTION_NOT_NEEDED'
+    description = 'Leader election not needed for topic partition.'
+    retriable = True
+    invalid_metadata = True
+
+
+class NoReassignmentInProgressError(BrokerResponseError):
+    errno = 85
+    message = 'NO_REASSIGNMENT_IN_PROGRESS'
+    description = 'No partition reassignment is in progress.'
+    retriable = False
+
+
+class GroupSubscribedToTopicError(BrokerResponseError):
+    errno = 86
+    message = 'GROUP_SUBSCRIBED_TO_TOPIC'
+    description = 'Deleting offsets of a topic is forbidden while the consumer group is actively subscribed to it.'
+    retriable = False
+
+
+class InvalidRecordError(BrokerResponseError):
+    errno = 87
+    message = 'INVALID_RECORD'
+    description = 'This record has failed the validation on broker and hence will be rejected.'
+    retriable = False
+
+
+class UnstableOffsetCommitError(BrokerResponseError):
+    errno = 88
+    message = 'UNSTABLE_OFFSET_COMMIT'
+    description = 'There are unstable offsets that need to be cleared.'
+    retriable = True
+
+
+class ThrottlingQuotaExceededError(BrokerResponseError):
+    errno = 89
+    message = 'THROTTLING_QUOTA_EXCEEDED'
+    description = 'The throttling quota has been exceeded.'
+    retriable = True
+
+
+class ProducerFencedError(BrokerResponseError):
+    errno = 90
+    message = 'PRODUCER_FENCED'
+    description = 'There is a newer producer with the same transactionalId which fences the current one.'
+    retriable = False
+
+
+class ResourceNotFoundError(BrokerResponseError):
+    errno = 91
+    message = 'RESOURCE_NOT_FOUND'
+    description = 'A request illegally referred to a resource that does not exist.'
+    retriable = False
+
+
+class DuplicateResourceError(BrokerResponseError):
+    errno = 92
+    message = 'DUPLICATE_RESOURCE'
+    description = 'A request illegally referred to the same resource twice.'
+    retriable = False
+
+
+class UnacceptableCredentialError(BrokerResponseError):
+    errno = 93
+    message = 'UNACCEPTABLE_CREDENTIAL'
+    description = 'Requested credential would not meet criteria for acceptability.'
+    retriable = False
+
+
+class InconsistentVoterSetError(BrokerResponseError):
+    errno = 94
+    message = 'INCONSISTENT_VOTER_SET'
+    description = 'Indicates that the either the sender or recipient of a voter-only request is not one of the expected voters.'
+    retriable = False
+
+
+class InvalidUpdateVersionError(BrokerResponseError):
+    errno = 95
+    message = 'INVALID_UPDATE_VERSION'
+    description = 'The given update version was invalid.'
+    retriable = False
+
+
+class FeatureUpdateFailedError(BrokerResponseError):
+    errno = 96
+    message = 'FEATURE_UPDATE_FAILED'
+    description = 'Unable to update finalized features due to an unexpected server error.'
+    retriable = False
+
+
+class PrincipalDeserializationFailureError(BrokerResponseError):
+    errno = 97
+    message = 'PRINCIPAL_DESERIALIZATION_FAILURE'
+    description = 'Request principal deserialization failed during forwarding. This indicates an internal error on the broker cluster security setup.'
+    retriable = False
+
+
+class SnapshotNotFoundError(BrokerResponseError):
+    errno = 98
+    message = 'SNAPSHOT_NOT_FOUND'
+    description = 'Requested snapshot was not found.'
+    retriable = False
+
+
+class PositionOutOfRangeError(BrokerResponseError):
+    errno = 99
+    message = 'POSITION_OUT_OF_RANGE'
+    description = 'Requested position is not greater than or equal to zero, and less than the size of the snapshot.'
+    retriable = False
+
+
+class UnknownTopicIdError(BrokerResponseError):
+    errno = 100
+    message = 'UNKNOWN_TOPIC_ID'
+    description = 'This server does not host this topic ID.'
+    retriable = True
+    invalid_metadata = True
+
+
+class DuplicateBrokerRegistrationError(BrokerResponseError):
+    errno = 101
+    message = 'DUPLICATE_BROKER_REGISTRATION'
+    description = 'This broker ID is already in use.'
+    retriable = False
+
+
+class BrokerIdNotRegisteredError(BrokerResponseError):
+    errno = 102
+    message = 'BROKER_ID_NOT_REGISTERED'
+    description = 'The given broker ID was not registered.'
+    retriable = False
+
+
+class InconsistentTopicIdError(BrokerResponseError):
+    errno = 103
+    message = 'INCONSISTENT_TOPIC_ID'
+    description = 'The log\'s topic ID did not match the topic ID in the request.'
+    retriable = True
+    invalid_metadata = True
+
+
+class InconsistentClusterIdError(BrokerResponseError):
+    errno = 104
+    message = 'INCONSISTENT_CLUSTER_ID'
+    description = 'The clusterId in the request does not match that found on the server.'
+    retriable = False
+
+
+class TransactionalIdNotFoundError(BrokerResponseError):
+    errno = 105
+    message = 'TRANSACTIONAL_ID_NOT_FOUND'
+    description = 'The transactionalId could not be found.'
+    retriable = False
+
+
+class FetchSessionTopicIdError(BrokerResponseError):
+    errno = 106
+    message = 'FETCH_SESSION_TOPIC_ID_ERROR'
+    description = 'The fetch session encountered inconsistent topic ID usage.'
+    retriable = True
+
+
+class IneligibleReplicaError(BrokerResponseError):
+    errno = 107
+    message = 'INELIGIBLE_REPLICA'
+    description = 'The new ISR contains at least one ineligible replica.'
+    retriable = False
+
+
+class NewLeaderElectedError(BrokerResponseError):
+    errno = 108
+    message = 'NEW_LEADER_ELECTED'
+    description = 'The AlterPartition request successfully updated the partition state but the leader has changed.'
+    retriable = False
+
+
+class OffsetMovedToTieredStorageError(BrokerResponseError):
+    errno = 109
+    message = 'OFFSET_MOVED_TO_TIERED_STORAGE'
+    description = 'The requested offset is moved to tiered storage.'
+    retriable = False
+
+
+class FencedMemberEpochError(BrokerResponseError):
+    errno = 110
+    message = 'FENCED_MEMBER_EPOCH'
+    description = 'The member epoch is fenced by the group coordinator. The member must abandon all its partitions and rejoin.'
+    retriable = False
+
+
+class UnreleasedInstanceIdError(BrokerResponseError):
+    errno = 111
+    message = 'UNRELEASED_INSTANCE_ID'
+    description = 'The instance ID is still used by another member in the consumer group. That member must leave first.'
+    retriable = False
+
+
+class UnsupportedAssignorError(BrokerResponseError):
+    errno = 112
+    message = 'UNSUPPORTED_ASSIGNOR'
+    description = 'The assignor or its version range is not supported by the consumer group.'
+    retriable = False
+
+
+class StaleMemberEpochError(BrokerResponseError):
+    errno = 113
+    message = 'STALE_MEMBER_EPOCH'
+    description = 'The member epoch is stale. The member must retry after receiving its updated member epoch via the ConsumerGroupHeartbeat API.'
+    retriable = False
+
+
+class MismatchedEndpointTypeError(BrokerResponseError):
+    errno = 114
+    message = 'MISMATCHED_ENDPOINT_TYPE'
+    description = 'The request was sent to an endpoint of the wrong type.'
+    retriable = False
+
+
+class UnsupportedEndpointTypeError(BrokerResponseError):
+    errno = 115
+    message = 'UNSUPPORTED_ENDPOINT_TYPE'
+    description = 'This endpoint type is not supported yet.'
+    retriable = False
+
+
+class UnknownControllerIdError(BrokerResponseError):
+    errno = 116
+    message = 'UNKNOWN_CONTROLLER_ID'
+    description = 'This controller ID is not known.'
+    retriable = False
+
+
+class UnknownSubscriptionIdError(BrokerResponseError):
+    errno = 117
+    message = 'UNKNOWN_SUBSCRIPTION_ID'
+    description = 'Client sent a push telemetry request with an invalid or outdated subscription ID.'
+    retriable = False
+
+
+class TelemetryTooLargeError(BrokerResponseError):
+    errno = 118
+    message = 'TELEMETRY_TOO_LARGE'
+    description = 'Client sent a push telemetry request larger than the maximum size the broker will accept.'
+    retriable = False
+
+
+class InvalidRegistrationError(BrokerResponseError):
+    errno = 119
+    message = 'INVALID_REGISTRATION'
+    description = 'The controller has considered the broker registration to be invalid.'
+    retriable = False
+
+
+class TransactionAbortableError(BrokerResponseError):
+    errno = 120
+    message = 'TRANSACTION_ABORTABLE'
+    description = 'The server encountered an error with the transaction. The client can abort the transaction to continue using this transactional ID.'
+    retriable = False
+
+
+class InvalidRecordStateError(BrokerResponseError):
+    errno = 121
+    message = 'INVALID_RECORD_STATE'
+    description = 'The record state is invalid. The acknowledgement of delivery could not be completed.'
+    retriable = False
+
+
+class ShareSessionNotFoundError(BrokerResponseError):
+    errno = 122
+    message = 'SHARE_SESSION_NOT_FOUND'
+    description = 'The share session was not found.'
+    retriable = True
+
+
+class InvalidShareSessionEpochError(BrokerResponseError):
+    errno = 123
+    message = 'INVALID_SHARE_SESSION_EPOCH'
+    description = 'The share session epoch is invalid.'
+    retriable = True
+
+
+class FencedStateEpochError(BrokerResponseError):
+    errno = 124
+    message = 'FENCED_STATE_EPOCH'
+    description = 'The share coordinator rejected the request because the share-group state epoch did not match.'
+    retriable = False
+
+
+class InvalidVoterKeyError(BrokerResponseError):
+    errno = 125
+    message = 'INVALID_VOTER_KEY'
+    description = 'The voter key doesn\'t match the receiving replica\'s key.'
+    retriable = False
+
+
+class DuplicateVoterError(BrokerResponseError):
+    errno = 126
+    message = 'DUPLICATE_VOTER'
+    description = 'The voter is already part of the set of voters.'
+    retriable = False
+
+
+class VoterNotFoundError(BrokerResponseError):
+    errno = 127
+    message = 'VOTER_NOT_FOUND'
+    description = 'The voter is not part of the set of voters.'
+    retriable = False
 
 
 class KafkaUnavailableError(KafkaError):
@@ -512,27 +1086,12 @@ kafka_errors = dict([(x.errno, x) for x in _iter_broker_errors()])
 
 
 def for_code(error_code):
-    return kafka_errors.get(error_code, UnknownError)
-
-
-def check_error(response):
-    if isinstance(response, Exception):
-        raise response
-    if response.error:
-        error_class = kafka_errors.get(response.error, UnknownError)
-        raise error_class(response)
-
-
-RETRY_BACKOFF_ERROR_TYPES = (
-    KafkaUnavailableError, LeaderNotAvailableError,
-    KafkaConnectionError, FailedPayloadsError
-)
-
-
-RETRY_REFRESH_ERROR_TYPES = (
-    NotLeaderForPartitionError, UnknownTopicOrPartitionError,
-    LeaderNotAvailableError, KafkaConnectionError
-)
-
-
-RETRY_ERROR_TYPES = RETRY_BACKOFF_ERROR_TYPES + RETRY_REFRESH_ERROR_TYPES
+    if error_code in kafka_errors:
+        return kafka_errors[error_code]
+    else:
+        # The broker error code was not found in our list. This can happen when connecting
+        # to a newer broker (with new error codes), or simply because our error list is
+        # not complete.
+        #
+        # To avoid dropping the error code, create a dynamic error class w/ errno override.
+        return type('UnrecognizedBrokerError', (UnknownError,), {'errno': error_code})
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index dd1cc50..155e9ee 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -216,7 +216,7 @@ class KafkaProducer(object):
             reconnection attempts will continue periodically with this fixed
             rate. To avoid connection storms, a randomization factor of 0.2
             will be applied to the backoff resulting in a random range between
-            20% below and 20% above the computed value. Default: 1000.
+            20% below and 20% above the computed value. Default: 30000.
         max_in_flight_requests_per_connection (int): Requests are pipelined
             to kafka brokers up to this number of maximum requests per
             broker connection. Note that if this setting is set to be greater
@@ -311,7 +311,7 @@ class KafkaProducer(object):
         'sock_chunk_bytes': 4096,  # undocumented experimental option
         'sock_chunk_buffer_count': 1000,  # undocumented experimental option
         'reconnect_backoff_ms': 50,
-        'reconnect_backoff_max_ms': 1000,
+        'reconnect_backoff_max_ms': 30000,
         'max_in_flight_requests_per_connection': 5,
         'security_protocol': 'PLAINTEXT',
         'ssl_context': None,
@@ -615,7 +615,7 @@ class KafkaProducer(object):
             # for API exceptions return them in the future,
             # for other exceptions raise directly
         except Errors.BrokerResponseError as e:
-            log.debug("Exception occurred during message send: %s", e)
+            log.error("Exception occurred during message send: %s", e)
             return FutureRecordMetadata(
                 FutureProduceResult(TopicPartition(topic, partition)),
                 -1, None, None,
diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py
index 35688d3..c6cd76c 100644
--- a/kafka/producer/sender.py
+++ b/kafka/producer/sender.py
@@ -103,14 +103,14 @@ class Sender(threading.Thread):
             self._metadata.request_update()
 
         # remove any nodes we aren't ready to send to
-        not_ready_timeout = float('inf')
+        not_ready_timeout_ms = float('inf')
         for node in list(ready_nodes):
             if not self._client.is_ready(node):
-                log.debug('Node %s not ready; delaying produce of accumulated batch', node)
+                node_delay_ms = self._client.connection_delay(node)
+                log.debug('Node %s not ready; delaying produce of accumulated batch (%f ms)', node, node_delay_ms)
                 self._client.maybe_connect(node, wakeup=False)
                 ready_nodes.remove(node)
-                not_ready_timeout = min(not_ready_timeout,
-                                        self._client.connection_delay(node))
+                not_ready_timeout_ms = min(not_ready_timeout_ms, node_delay_ms)
 
         # create produce requests
         batches_by_node = self._accumulator.drain(
@@ -136,7 +136,7 @@ class Sender(threading.Thread):
         # off). Note that this specifically does not include nodes with
         # sendable data that aren't ready to send since they would cause busy
         # looping.
-        poll_timeout_ms = min(next_ready_check_delay * 1000, not_ready_timeout)
+        poll_timeout_ms = min(next_ready_check_delay * 1000, not_ready_timeout_ms)
         if ready_nodes:
             log.debug("Nodes with data ready to send: %s", ready_nodes) # trace
             log.debug("Created %d produce requests: %s", len(requests), requests) # trace
@@ -181,7 +181,7 @@ class Sender(threading.Thread):
             self.wakeup()
 
     def _failed_produce(self, batches, node_id, error):
-        log.debug("Error sending produce request to node %d: %s", node_id, error) # trace
+        log.error("Error sending produce request to node %d: %s", node_id, error) # trace
         for batch in batches:
             self._complete_batch(batch, error, -1, None)
 
diff --git a/servers/resources/default/kafka.properties b/servers/resources/default/kafka.properties
new file mode 100644
index 0000000..71b20f5
--- /dev/null
+++ b/servers/resources/default/kafka.properties
@@ -0,0 +1,171 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# see kafka.server.KafkaConfig for additional details and defaults
+
+############################# Server Basics #############################
+
+# The id of the broker. This must be set to a unique integer for each broker.
+broker.id={broker_id}
+
+############################# Socket Server Settings #############################
+
+# The address the socket server listens on. It will get the value returned from 
+# java.net.InetAddress.getCanonicalHostName() if not configured.
+#   FORMAT:
+#     listeners = listener_name://host_name:port
+#   EXAMPLE:
+#     listeners = PLAINTEXT://your.host.name:9092
+listeners={transport}://{host}:{port}
+security.inter.broker.protocol={transport}
+
+{sasl_config}
+
+ssl.keystore.location={ssl_dir}/kafka.server.keystore.jks
+ssl.keystore.password=foobar
+ssl.key.password=foobar
+ssl.truststore.location={ssl_dir}/kafka.server.truststore.jks
+ssl.truststore.password=foobar
+
+authorizer.class.name=kafka.security.authorizer.AclAuthorizer
+allow.everyone.if.no.acl.found=true
+
+# The port the socket server listens on
+#port=9092
+
+# Hostname and port the broker will advertise to producers and consumers. If not set, 
+# it uses the value for "listeners" if configured.  Otherwise, it will use the value
+# returned from java.net.InetAddress.getCanonicalHostName().
+#advertised.listeners=PLAINTEXT://your.host.name:9092
+
+# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
+#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
+
+# The number of threads that the server uses for receiving requests from the network and sending responses to the network
+num.network.threads=3
+
+# The number of threads that the server uses for processing requests, which may include disk I/O
+num.io.threads=8
+
+# The send buffer (SO_SNDBUF) used by the socket server
+socket.send.buffer.bytes=102400
+
+# The receive buffer (SO_RCVBUF) used by the socket server
+socket.receive.buffer.bytes=102400
+
+# The maximum size of a request that the socket server will accept (protection against OOM)
+socket.request.max.bytes=104857600
+
+
+############################# Log Basics #############################
+
+# A comma separated list of directories under which to store log files
+log.dirs={tmp_dir}/data
+
+# The default number of log partitions per topic. More partitions allow greater
+# parallelism for consumption, but this will also result in more files across
+# the brokers.
+num.partitions={partitions}
+default.replication.factor={replicas}
+
+## Short Replica Lag -- Drops failed brokers out of ISR
+replica.lag.time.max.ms=1000
+replica.socket.timeout.ms=1000
+
+# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
+# This value is recommended to be increased for installations with data dirs located in RAID array.
+num.recovery.threads.per.data.dir=1
+
+############################# Internal Topic Settings  #############################
+# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
+# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
+offsets.topic.replication.factor=1
+transaction.state.log.replication.factor=1
+transaction.state.log.min.isr=1
+
+############################# Log Flush Policy #############################
+
+# Messages are immediately written to the filesystem but by default we only fsync() to sync
+# the OS cache lazily. The following configurations control the flush of data to disk.
+# There are a few important trade-offs here:
+#    1. Durability: Unflushed data may be lost if you are not using replication.
+#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
+#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
+# The settings below allow one to configure the flush policy to flush data after a period of time or
+# every N messages (or both). This can be done globally and overridden on a per-topic basis.
+
+# The number of messages to accept before forcing a flush of data to disk
+#log.flush.interval.messages=10000
+
+# The maximum amount of time a message can sit in a log before we force a flush
+#log.flush.interval.ms=1000
+
+############################# Log Retention Policy #############################
+
+# The following configurations control the disposal of log segments. The policy can
+# be set to delete segments after a period of time, or after a given size has accumulated.
+# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
+# from the end of the log.
+
+# The minimum age of a log file to be eligible for deletion due to age
+log.retention.hours=168
+
+# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
+# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
+#log.retention.bytes=1073741824
+
+# The maximum size of a log segment file. When this size is reached a new log segment will be created.
+log.segment.bytes=1073741824
+
+# The interval at which log segments are checked to see if they can be deleted according
+# to the retention policies
+log.retention.check.interval.ms=300000
+
+# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
+# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
+log.cleaner.enable=false
+
+# tune down offset topics to reduce setup time in tests
+offsets.commit.timeout.ms=500
+offsets.topic.num.partitions=2
+offsets.topic.replication.factor=1
+
+# Allow shorter session timeouts for tests
+group.min.session.timeout.ms=1000
+
+
+############################# Zookeeper #############################
+
+# Zookeeper connection string (see zookeeper docs for details).
+# This is a comma separated host:port pairs, each corresponding to a zk
+# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
+# You can also append an optional chroot string to the urls to specify the
+# root directory for all kafka znodes.
+zookeeper.connect={zk_host}:{zk_port}/{zk_chroot}
+
+# Timeout in ms for connecting to zookeeper
+zookeeper.connection.timeout.ms=30000
+# We want to expire kafka broker sessions quickly when brokers die b/c we restart them quickly
+zookeeper.session.timeout.ms=500
+
+
+############################# Group Coordinator Settings #############################
+
+# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
+# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
+# The default value for this is 3 seconds.
+# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
+# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
+group.initial.rebalance.delay.ms=0
diff --git a/servers/resources/default/kafka_server_jaas.conf b/servers/resources/default/kafka_server_jaas.conf
new file mode 100644
index 0000000..18efe43
--- /dev/null
+++ b/servers/resources/default/kafka_server_jaas.conf
@@ -0,0 +1,4 @@
+KafkaServer {{
+    {jaas_config}
+}};
+Client {{}};
\ No newline at end of file
diff --git a/servers/resources/default/log4j.properties b/servers/resources/default/log4j.properties
new file mode 100644
index 0000000..b0b76aa
--- /dev/null
+++ b/servers/resources/default/log4j.properties
@@ -0,0 +1,25 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+# 
+#    http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+log4j.rootLogger=INFO, stdout, logfile
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.appender.logfile=org.apache.log4j.FileAppender
+log4j.appender.logfile.File=${kafka.logs.dir}/server.log
+log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
+log4j.appender.logfile.layout.ConversionPattern=[%d] %p %m (%c)%n
diff --git a/servers/resources/default/zookeeper.properties b/servers/resources/default/zookeeper.properties
new file mode 100644
index 0000000..e3fd097
--- /dev/null
+++ b/servers/resources/default/zookeeper.properties
@@ -0,0 +1,21 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+# 
+#    http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# the directory where the snapshot is stored.
+dataDir={tmp_dir}
+# the port at which the clients will connect
+clientPort={port}
+clientPortAddress={host}
+# disable the per-ip limit on the number of connections since this is a non-production config
+maxClientCnxns=0


Test patch:
diff --git a/test/conftest.py b/test/conftest.py
index 3fa0262..d54a912 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -137,7 +137,9 @@ def conn(mocker):
         MetadataResponse[0](
             [(0, 'foo', 12), (1, 'bar', 34)],  # brokers
             []))  # topics
+    conn.connection_delay.return_value = 0
     conn.blacked_out.return_value = False
+    conn.next_ifr_request_timeout_ms.return_value = float('inf')
     def _set_conn_state(state):
         conn.state = state
         return state
diff --git a/test/fixtures.py b/test/fixtures.py
index 8ae25dd..38ea6f0 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -111,7 +111,10 @@ class Fixture(object):
 
     @classmethod
     def test_resource(cls, filename):
-        return os.path.join(cls.project_root, "servers", cls.kafka_version, "resources", filename)
+        path = os.path.join(cls.project_root, "servers", cls.kafka_version, "resources", filename)
+        if os.path.isfile(path):
+            return path
+        return os.path.join(cls.project_root, "servers", "resources", "default", filename)
 
     @classmethod
     def kafka_run_class_args(cls, *args):
diff --git a/test/test_client_async.py b/test/test_client_async.py
index 66b227a..ec5e2c0 100644
--- a/test/test_client_async.py
+++ b/test/test_client_async.py
@@ -230,29 +230,25 @@ def test_send(cli, conn):
 
 def test_poll(mocker):
     metadata = mocker.patch.object(KafkaClient, '_maybe_refresh_metadata')
+    ifr_request_timeout = mocker.patch.object(KafkaClient, '_next_ifr_request_timeout_ms')
     _poll = mocker.patch.object(KafkaClient, '_poll')
-    ifrs = mocker.patch.object(KafkaClient, 'in_flight_request_count')
-    ifrs.return_value = 1
     cli = KafkaClient(api_version=(0, 9))
 
     # metadata timeout wins
+    ifr_request_timeout.return_value = float('inf')
     metadata.return_value = 1000
     cli.poll()
     _poll.assert_called_with(1.0)
 
     # user timeout wins
-    cli.poll(250)
+    cli.poll(timeout_ms=250)
     _poll.assert_called_with(0.25)
 
-    # default is request_timeout_ms
+    # ifr request timeout wins
+    ifr_request_timeout.return_value = 30000
     metadata.return_value = 1000000
     cli.poll()
-    _poll.assert_called_with(cli.config['request_timeout_ms'] / 1000.0)
-
-    # If no in-flight-requests, drop timeout to retry_backoff_ms
-    ifrs.return_value = 0
-    cli.poll()
-    _poll.assert_called_with(cli.config['retry_backoff_ms'] / 1000.0)
+    _poll.assert_called_with(30.0)
 
 
 def test__poll():
@@ -309,25 +305,24 @@ def client(mocker):
 
 def test_maybe_refresh_metadata_ttl(mocker, client):
     client.cluster.ttl.return_value = 1234
-    mocker.patch.object(KafkaClient, 'in_flight_request_count', return_value=1)
 
     client.poll(timeout_ms=12345678)
     client._poll.assert_called_with(1.234)
 
 
 def test_maybe_refresh_metadata_backoff(mocker, client):
-    mocker.patch.object(KafkaClient, 'in_flight_request_count', return_value=1)
+    mocker.patch.object(client, 'least_loaded_node', return_value=None)
+    mocker.patch.object(client, 'least_loaded_node_refresh_ms', return_value=4321)
     now = time.time()
     t = mocker.patch('time.time')
     t.return_value = now
 
     client.poll(timeout_ms=12345678)
-    client._poll.assert_called_with(2.222) # reconnect backoff
+    client._poll.assert_called_with(4.321)
 
 
 def test_maybe_refresh_metadata_in_progress(mocker, client):
     client._metadata_refresh_in_progress = True
-    mocker.patch.object(KafkaClient, 'in_flight_request_count', return_value=1)
 
     client.poll(timeout_ms=12345678)
     client._poll.assert_called_with(9999.999) # request_timeout_ms
@@ -336,7 +331,6 @@ def test_maybe_refresh_metadata_in_progress(mocker, client):
 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)
-    mocker.patch.object(KafkaClient, 'in_flight_request_count', return_value=1)
     send = mocker.patch.object(client, 'send')
 
     client.poll(timeout_ms=12345678)
@@ -348,10 +342,10 @@ def test_maybe_refresh_metadata_update(mocker, client):
 
 def test_maybe_refresh_metadata_cant_send(mocker, client):
     mocker.patch.object(client, 'least_loaded_node', return_value='foobar')
+    mocker.patch.object(client, '_can_send_request', return_value=False)
     mocker.patch.object(client, '_can_connect', return_value=True)
     mocker.patch.object(client, '_maybe_connect', return_value=True)
     mocker.patch.object(client, 'maybe_connect', return_value=True)
-    mocker.patch.object(KafkaClient, 'in_flight_request_count', return_value=1)
 
     now = time.time()
     t = mocker.patch('time.time')
@@ -359,14 +353,14 @@ def test_maybe_refresh_metadata_cant_send(mocker, client):
 
     # first poll attempts connection
     client.poll(timeout_ms=12345678)
-    client._poll.assert_called_with(2.222) # reconnect backoff
+    client._poll.assert_called_with(12345.678)
     client.maybe_connect.assert_called_once_with('foobar', wakeup=False)
 
     # poll while connecting should not attempt a new connection
     client._connecting.add('foobar')
     client._can_connect.reset_mock()
     client.poll(timeout_ms=12345678)
-    client._poll.assert_called_with(2.222) # connection timeout (reconnect timeout)
+    client._poll.assert_called_with(12345.678)
     assert not client._can_connect.called
 
     assert not client._metadata_refresh_in_progress
diff --git a/test/test_codec.py b/test/test_codec.py
index e057074..24159c2 100644
--- a/test/test_codec.py
+++ b/test/test_codec.py
@@ -39,12 +39,14 @@ def test_snappy_detect_xerial():
     _detect_xerial_stream = kafka1.codec._detect_xerial_stream
 
     header = b'\x82SNAPPY\x00\x00\x00\x00\x01\x00\x00\x00\x01Some extra bytes'
+    redpanda_header = b'\x82SNAPPY\x00\x01\x00\x00\x00\x01\x00\x00\x00Some extra bytes'
     false_header = b'\x01SNAPPY\x00\x00\x00\x01\x00\x00\x00\x01'
     default_snappy = snappy_encode(b'foobar' * 50)
     random_snappy = snappy_encode(b'SNAPPY' * 50, xerial_compatible=False)
     short_data = b'\x01\x02\x03\x04'
 
     assert _detect_xerial_stream(header) is True
+    assert _detect_xerial_stream(redpanda_header) is True
     assert _detect_xerial_stream(b'') is False
     assert _detect_xerial_stream(b'\x00') is False
     assert _detect_xerial_stream(false_header) is False
diff --git a/test/test_conn.py b/test/test_conn.py
index 966f7b3..3afa942 100644
--- a/test/test_conn.py
+++ b/test/test_conn.py
@@ -80,15 +80,35 @@ def test_blacked_out(conn):
         assert conn.blacked_out() is True
 
 
-def test_connection_delay(conn):
+def test_connection_delay(conn, mocker):
+    mocker.patch.object(conn, '_reconnect_jitter_pct', return_value=1.0)
     with mock.patch("time.time", return_value=1000):
         conn.last_attempt = 1000
         assert conn.connection_delay() == conn.config['reconnect_backoff_ms']
         conn.state = ConnectionStates.CONNECTING
-        assert conn.connection_delay() == float('inf')
+        assert conn.connection_delay() == conn.config['reconnect_backoff_ms']
         conn.state = ConnectionStates.CONNECTED
         assert conn.connection_delay() == float('inf')
 
+        conn._gai.clear()
+        conn._update_reconnect_backoff()
+        conn.state = ConnectionStates.DISCONNECTED
+        assert conn.connection_delay() == 1.0 * conn.config['reconnect_backoff_ms']
+        conn.state = ConnectionStates.CONNECTING
+        assert conn.connection_delay() == 1.0 * conn.config['reconnect_backoff_ms']
+
+        conn._update_reconnect_backoff()
+        conn.state = ConnectionStates.DISCONNECTED
+        assert conn.connection_delay() == 2.0 * conn.config['reconnect_backoff_ms']
+        conn.state = ConnectionStates.CONNECTING
+        assert conn.connection_delay() == 2.0 * conn.config['reconnect_backoff_ms']
+
+        conn._update_reconnect_backoff()
+        conn.state = ConnectionStates.DISCONNECTED
+        assert conn.connection_delay() == 4.0 * conn.config['reconnect_backoff_ms']
+        conn.state = ConnectionStates.CONNECTING
+        assert conn.connection_delay() == 4.0 * conn.config['reconnect_backoff_ms']
+
 
 def test_connected(conn):
     assert conn.connected() is False