Repository: dpkp/kafka-python
Pull request #2585 (2025-04-07 21:32:00)

Rename Coordinator errors to generic not just consumer group


Patch:
diff --git a/kafka/admin/client.py b/kafka/admin/client.py
index 392687b..94de5a8 100644
--- a/kafka/admin/client.py
+++ b/kafka/admin/client.py
@@ -1460,9 +1460,9 @@ class KafkaAdminClient(object):
             list: List of tuples of Consumer Groups.
 
         Raises:
-            GroupCoordinatorNotAvailableError: The coordinator is not
+            CoordinatorNotAvailableError: The coordinator is not
                 available, so cannot process requests.
-            GroupLoadInProgressError: The coordinator is loading and
+            CoordinatorLoadInProgressError: The coordinator is loading and
                 hence can't process requests.
         """
         # While we return a list, internally use a set to prevent duplicates
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index 508e35a..ed0c50a 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -12,7 +12,7 @@ from kafka.vendor import six
 import kafka.errors as Errors
 from kafka.future import Future
 from kafka.metrics.stats import Avg, Count, Max, Rate
-from kafka.protocol.fetch import FetchRequest
+from kafka.protocol.fetch import FetchRequest, AbortedTransaction
 from kafka.protocol.list_offsets import (
     ListOffsetsRequest, OffsetResetStrategy, UNKNOWN_OFFSET
 )
@@ -28,6 +28,11 @@ log = logging.getLogger(__name__)
 READ_UNCOMMITTED = 0
 READ_COMMITTED = 1
 
+ISOLATION_LEVEL_CONFIG = {
+    'read_uncommitted': READ_UNCOMMITTED,
+    'read_committed': READ_COMMITTED,
+}
+
 ConsumerRecord = collections.namedtuple("ConsumerRecord",
     ["topic", "partition", "leader_epoch", "offset", "timestamp", "timestamp_type",
      "key", "value", "headers", "checksum", "serialized_key_size", "serialized_value_size", "serialized_header_size"])
@@ -60,6 +65,7 @@ class Fetcher(six.Iterator):
         'metric_group_prefix': 'consumer',
         'retry_backoff_ms': 100,
         'enable_incremental_fetch_sessions': True,
+        'isolation_level': 'read_uncommitted',
     }
 
     def __init__(self, client, subscriptions, **configs):
@@ -100,12 +106,18 @@ class Fetcher(six.Iterator):
                 consumed. This ensures no on-the-wire or on-disk corruption to
                 the messages occurred. This check adds some overhead, so it may
                 be disabled in cases seeking extreme performance. Default: True
+            isolation_level (str): Configure KIP-98 transactional consumer by
+                setting to 'read_committed'. This will cause the consumer to
+                skip records from aborted tranactions. Default: 'read_uncommitted'
         """
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
             if key in configs:
                 self.config[key] = configs[key]
 
+        if self.config['isolation_level'] not in ISOLATION_LEVEL_CONFIG:
+            raise Errors.KafkaConfigurationError('Unrecognized isolation_level')
+
         self._client = client
         self._subscriptions = subscriptions
         self._completed_fetches = collections.deque()  # Unparsed responses
@@ -116,7 +128,7 @@ class Fetcher(six.Iterator):
             self._sensors = FetchManagerMetrics(self.config['metrics'], self.config['metric_group_prefix'])
         else:
             self._sensors = None
-        self._isolation_level = READ_UNCOMMITTED
+        self._isolation_level = ISOLATION_LEVEL_CONFIG[self.config['isolation_level']]
         self._session_handlers = {}
         self._nodes_with_pending_fetch_requests = set()
 
@@ -244,7 +256,7 @@ class Fetcher(six.Iterator):
         else:
             raise NoOffsetForPartitionError(partition)
 
-        log.debug("Resetting offset for partition %s to %s offset.",
+        log.debug("Resetting offset for partition %s to offset %s.",
                   partition, strategy)
         offsets = self._retrieve_offsets({partition: timestamp}, timeout_ms=timeout_ms)
 
@@ -765,14 +777,21 @@ class Fetcher(six.Iterator):
                     return None
 
                 records = MemoryRecords(completed_fetch.partition_data[-1])
+                aborted_transactions = None
+                if completed_fetch.response_version >= 11:
+                    aborted_transactions = completed_fetch.partition_data[-3]
+                elif completed_fetch.response_version >= 4:
+                    aborted_transactions = completed_fetch.partition_data[-2]
                 log.debug("Preparing to read %s bytes of data for partition %s with offset %d",
                           records.size_in_bytes(), tp, fetch_offset)
                 parsed_records = self.PartitionRecords(fetch_offset, tp, records,
-                                                       self.config['key_deserializer'],
-                                                       self.config['value_deserializer'],
-                                                       self.config['check_crcs'],
-                                                       completed_fetch.metric_aggregator,
-                                                       self._on_partition_records_drain)
+                                                       key_deserializer=self.config['key_deserializer'],
+                                                       value_deserializer=self.config['value_deserializer'],
+                                                       check_crcs=self.config['check_crcs'],
+                                                       isolation_level=self._isolation_level,
+                                                       aborted_transactions=aborted_transactions,
+                                                       metric_aggregator=completed_fetch.metric_aggregator,
+                                                       on_drain=self._on_partition_records_drain)
                 if not records.has_next() and records.size_in_bytes() > 0:
                     if completed_fetch.response_version < 3:
                         # Implement the pre KIP-74 behavior of throwing a RecordTooLargeException.
@@ -845,13 +864,23 @@ class Fetcher(six.Iterator):
             self._next_partition_records.drain()
 
     class PartitionRecords(object):
-        def __init__(self, fetch_offset, tp, records, key_deserializer, value_deserializer, check_crcs, metric_aggregator, on_drain):
+        def __init__(self, fetch_offset, tp, records,
+                     key_deserializer=None, value_deserializer=None,
+                     check_crcs=True, isolation_level=READ_UNCOMMITTED,
+                     aborted_transactions=None, # raw data from response / list of (producer_id, first_offset) tuples
+                     metric_aggregator=None, on_drain=lambda x: None):
             self.fetch_offset = fetch_offset
             self.topic_partition = tp
             self.leader_epoch = -1
             self.next_fetch_offset = fetch_offset
             self.bytes_read = 0
             self.records_read = 0
+            self.isolation_level = isolation_level
+            self.aborted_producer_ids = set()
+            self.aborted_transactions = collections.deque(
+                sorted([AbortedTransaction(*data) for data in aborted_transactions] if aborted_transactions else [],
+                       key=lambda txn: txn.first_offset)
+            )
             self.metric_aggregator = metric_aggregator
             self.check_crcs = check_crcs
             self.record_iterator = itertools.dropwhile(
@@ -900,18 +929,35 @@ class Fetcher(six.Iterator):
                                 "Record batch for partition %s at offset %s failed crc check" % (
                                     self.topic_partition, batch.base_offset))
 
+
                     # Try DefaultsRecordBatch / message log format v2
-                    # base_offset, last_offset_delta, and control batches
+                    # base_offset, last_offset_delta, aborted transactions, and control batches
                     if batch.magic == 2:
                         self.leader_epoch = batch.leader_epoch
+                        if self.isolation_level == READ_COMMITTED and batch.has_producer_id():
+                            # remove from the aborted transaction queue all aborted transactions which have begun
+                            # before the current batch's last offset and add the associated producerIds to the
+                            # aborted producer set
+                            self._consume_aborted_transactions_up_to(batch.last_offset)
+
+                            producer_id = batch.producer_id
+                            if self._contains_abort_marker(batch):
+                                try:
+                                    self.aborted_producer_ids.remove(producer_id)
+                                except KeyError:
+                                    pass
+                            elif self._is_batch_aborted(batch):
+                                log.debug("Skipping aborted record batch from partition %s with producer_id %s and"
+                                          " offsets %s to %s",
+                                          self.topic_partition, producer_id, batch.base_offset, batch.last_offset)
+                                self.next_fetch_offset = batch.next_offset
+                                batch = records.next_batch()
+                                continue
+
                         # Control batches have a single record indicating whether a transaction
-                        # was aborted or committed.
-                        # When isolation_level is READ_COMMITTED (currently unsupported)
-                        # we should also skip all messages from aborted transactions
-                        # For now we only support READ_UNCOMMITTED and so we ignore the
-                        # abort/commit signal.
+                        # was aborted or committed. These are not returned to the consumer.
                         if batch.is_control_batch:
-                            self.next_fetch_offset = next(batch).offset + 1
+                            self.next_fetch_offset = batch.next_offset
                             batch = records.next_batch()
                             continue
 
@@ -944,7 +990,7 @@ class Fetcher(six.Iterator):
                     # unnecessary re-fetching of the same batch (in the worst case, the consumer could get stuck
                     # fetching the same batch repeatedly).
                     if last_batch and last_batch.magic == 2:
-                        self.next_fetch_offset = last_batch.base_offset + last_batch.last_offset_delta + 1
+                        self.next_fetch_offset = last_batch.next_offset
                     self.drain()
 
             # If unpacking raises StopIteration, it is erroneously
@@ -961,6 +1007,24 @@ class Fetcher(six.Iterator):
                 return f.deserialize(topic, bytes_)
             return f(bytes_)
 
+        def _consume_aborted_transactions_up_to(self, offset):
+            if not self.aborted_transactions:
+                return
+
+            while self.aborted_transactions and self.aborted_transactions[0].first_offset <= offset:
+                self.aborted_producer_ids.add(self.aborted_transactions.popleft().producer_id)
+
+        def _is_batch_aborted(self, batch):
+            return batch.is_transactional and batch.producer_id in self.aborted_producer_ids
+
+        def _contains_abort_marker(self, batch):
+            if not batch.is_control_batch:
+                return False
+            record = next(batch)
+            if not record:
+                return False
+            return record.abort
+
 
 class FetchSessionHandler(object):
     """
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 4a39dc1..7fff6e7 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -121,6 +121,9 @@ class KafkaConsumer(six.Iterator):
             consumed. This ensures no on-the-wire or on-disk corruption to
             the messages occurred. This check adds some overhead, so it may
             be disabled in cases seeking extreme performance. Default: True
+        isolation_level (str): Configure KIP-98 transactional consumer by
+            setting to 'read_committed'. This will cause the consumer to
+            skip records from aborted tranactions. Default: 'read_uncommitted'
         allow_auto_create_topics (bool): Enable/disable auto topic creation
             on metadata request. Only available with api_version >= (0, 11).
             Default: True
@@ -290,6 +293,7 @@ class KafkaConsumer(six.Iterator):
         'auto_commit_interval_ms': 5000,
         'default_offset_commit_callback': lambda offsets, response: True,
         'check_crcs': True,
+        'isolation_level': 'read_uncommitted',
         'allow_auto_create_topics': True,
         'metadata_max_age_ms': 5 * 60 * 1000,
         'partition_assignment_strategy': (RangePartitionAssignor, RoundRobinPartitionAssignor),
diff --git a/kafka/consumer/subscription_state.py b/kafka/consumer/subscription_state.py
index 4cc2102..0ff2ae9 100644
--- a/kafka/consumer/subscription_state.py
+++ b/kafka/consumer/subscription_state.py
@@ -1,7 +1,7 @@
 from __future__ import absolute_import
 
 import abc
-from collections import defaultdict, OrderedDict
+from collections import OrderedDict
 try:
     from collections.abc import Sequence
 except ImportError:
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index 0c238fd..4f413c7 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -478,7 +478,7 @@ class BaseCoordinator(object):
                 group leader
         """
         if self.coordinator_unknown():
-            e = Errors.GroupCoordinatorNotAvailableError(self.coordinator_id)
+            e = Errors.CoordinatorNotAvailableError(self.coordinator_id)
             return Future().failure(e)
 
         elif not self._client.ready(self.coordinator_id, metadata_priority=False):
@@ -555,7 +555,7 @@ class BaseCoordinator(object):
                 else:
                     self._on_join_follower().chain(future)
 
-        elif error_type is Errors.GroupLoadInProgressError:
+        elif error_type is Errors.CoordinatorLoadInProgressError:
             log.debug("Attempt to join group %s rejected since coordinator %s"
                       " is loading the group.", self.group_id, self.coordinator_id)
             # backoff and retry
@@ -567,8 +567,8 @@ class BaseCoordinator(object):
             log.debug("Attempt to join group %s failed due to unknown member id",
                       self.group_id)
             future.failure(error)
-        elif error_type in (Errors.GroupCoordinatorNotAvailableError,
-                            Errors.NotCoordinatorForGroupError):
+        elif error_type in (Errors.CoordinatorNotAvailableError,
+                            Errors.NotCoordinatorError):
             # re-discover the coordinator and retry with backoff
             self.coordinator_dead(error_type())
             log.debug("Attempt to join group %s failed due to obsolete "
@@ -636,7 +636,7 @@ class BaseCoordinator(object):
 
     def _send_sync_group_request(self, request):
         if self.coordinator_unknown():
-            e = Errors.GroupCoordinatorNotAvailableError(self.coordinator_id)
+            e = Errors.CoordinatorNotAvailableError(self.coordinator_id)
             return Future().failure(e)
 
         # We assume that coordinator is ready if we're sending SyncGroup
@@ -674,8 +674,8 @@ class BaseCoordinator(object):
             log.debug("SyncGroup for group %s failed due to %s", self.group_id, error)
             self.reset_generation()
             future.failure(error)
-        elif error_type in (Errors.GroupCoordinatorNotAvailableError,
-                            Errors.NotCoordinatorForGroupError):
+        elif error_type in (Errors.CoordinatorNotAvailableError,
+                            Errors.NotCoordinatorError):
             error = error_type()
             log.debug("SyncGroup for group %s failed due to %s", self.group_id, error)
             self.coordinator_dead(error)
@@ -732,7 +732,7 @@ class BaseCoordinator(object):
                 self.heartbeat.reset_timeouts()
             future.success(self.coordinator_id)
 
-        elif error_type is Errors.GroupCoordinatorNotAvailableError:
+        elif error_type is Errors.CoordinatorNotAvailableError:
             log.debug("Group Coordinator Not Available; retry")
             future.failure(error_type())
         elif error_type is Errors.GroupAuthorizationFailedError:
@@ -842,7 +842,7 @@ class BaseCoordinator(object):
     def _send_heartbeat_request(self):
         """Send a heartbeat request"""
         if self.coordinator_unknown():
-            e = Errors.GroupCoordinatorNotAvailableError(self.coordinator_id)
+            e = Errors.CoordinatorNotAvailableError(self.coordinator_id)
             return Future().failure(e)
 
         elif not self._client.ready(self.coordinator_id, metadata_priority=False):
@@ -869,8 +869,8 @@ class BaseCoordinator(object):
             log.debug("Received successful heartbeat response for group %s",
                       self.group_id)
             future.success(None)
-        elif error_type in (Errors.GroupCoordinatorNotAvailableError,
-                            Errors.NotCoordinatorForGroupError):
+        elif error_type in (Errors.CoordinatorNotAvailableError,
+                            Errors.NotCoordinatorError):
             log.warning("Heartbeat failed for group %s: coordinator (node %s)"
                         " is either not started or not valid", self.group_id,
                         self.coordinator())
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index 4bc7ba9..2944c7e 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -590,7 +590,7 @@ class ConsumerCoordinator(BaseCoordinator):
 
         node_id = self.coordinator()
         if node_id is None:
-            return Future().failure(Errors.GroupCoordinatorNotAvailableError)
+            return Future().failure(Errors.CoordinatorNotAvailableError)
 
 
         # create the offset commit request
@@ -719,14 +719,14 @@ class ConsumerCoordinator(BaseCoordinator):
                               " %s", self.group_id, tp, error_type.__name__)
                     future.failure(error_type())
                     return
-                elif error_type is Errors.GroupLoadInProgressError:
+                elif error_type is Errors.CoordinatorLoadInProgressError:
                     # just retry
                     log.debug("OffsetCommit for group %s failed: %s",
                               self.group_id, error_type.__name__)
                     future.failure(error_type(self.group_id))
                     return
-                elif error_type in (Errors.GroupCoordinatorNotAvailableError,
-                                    Errors.NotCoordinatorForGroupError,
+                elif error_type in (Errors.CoordinatorNotAvailableError,
+                                    Errors.NotCoordinatorError,
                                     Errors.RequestTimedOutError):
                     log.debug("OffsetCommit for group %s failed: %s",
                               self.group_id, error_type.__name__)
@@ -777,7 +777,7 @@ class ConsumerCoordinator(BaseCoordinator):
 
         node_id = self.coordinator()
         if node_id is None:
-            return Future().failure(Errors.GroupCoordinatorNotAvailableError)
+            return Future().failure(Errors.CoordinatorNotAvailableError)
 
         # Verify node is ready
         if not self._client.ready(node_id):
@@ -812,10 +812,10 @@ class ConsumerCoordinator(BaseCoordinator):
             error_type = Errors.for_code(response.error_code)
             log.debug("Offset fetch failed: %s", error_type.__name__)
             error = error_type()
-            if error_type is Errors.GroupLoadInProgressError:
+            if error_type is Errors.CoordinatorLoadInProgressError:
                 # Retry
                 future.failure(error)
-            elif error_type is Errors.NotCoordinatorForGroupError:
+            elif error_type is Errors.NotCoordinatorError:
                 # re-discover the coordinator and retry
                 self.coordinator_dead(error)
                 future.failure(error)
@@ -841,10 +841,10 @@ class ConsumerCoordinator(BaseCoordinator):
                     error = error_type()
                     log.debug("Group %s failed to fetch offset for partition"
                               " %s: %s", self.group_id, tp, error)
-                    if error_type is Errors.GroupLoadInProgressError:
+                    if error_type is Errors.CoordinatorLoadInProgressError:
                         # just retry
                         future.failure(error)
-                    elif error_type is Errors.NotCoordinatorForGroupError:
+                    elif error_type is Errors.NotCoordinatorError:
                         # re-discover the coordinator and retry
                         self.coordinator_dead(error)
                         future.failure(error)
diff --git a/kafka/errors.py b/kafka/errors.py
index aaba89d..76a9356 100644
--- a/kafka/errors.py
+++ b/kafka/errors.py
@@ -218,33 +218,28 @@ class NetworkExceptionError(BrokerResponseError):
     invalid_metadata = True
 
 
-class GroupLoadInProgressError(BrokerResponseError):
+class CoordinatorLoadInProgressError(BrokerResponseError):
     errno = 14
-    message = 'OFFSETS_LOAD_IN_PROGRESS'
-    description = ('The broker returns this error code for an offset fetch'
-                   ' request if it is still loading offsets (after a leader'
-                   ' change for that offsets topic partition), or in response'
-                   ' to group membership requests (such as heartbeats) when'
-                   ' group metadata is being loaded by the coordinator.')
+    message = 'COORDINATOR_LOAD_IN_PROGRESS'
+    description = ('The broker returns this error code for txn or group requests,'
+                   ' when the coordinator is loading and hence cant process requests')
     retriable = True
 
 
-class GroupCoordinatorNotAvailableError(BrokerResponseError):
+class CoordinatorNotAvailableError(BrokerResponseError):
     errno = 15
-    message = 'CONSUMER_COORDINATOR_NOT_AVAILABLE'
-    description = ('The broker returns this error code for group coordinator'
-                   ' requests, offset commits, and most group management'
+    message = 'COORDINATOR_NOT_AVAILABLE'
+    description = ('The broker returns this error code for consumer and transaction'
                    ' requests if the offsets topic has not yet been created, or'
-                   ' if the group coordinator is not active.')
+                   ' if the group/txn coordinator is not active.')
     retriable = True
 
 
-class NotCoordinatorForGroupError(BrokerResponseError):
+class NotCoordinatorError(BrokerResponseError):
     errno = 16
-    message = 'NOT_COORDINATOR_FOR_CONSUMER'
-    description = ('The broker returns this error code if it receives an offset'
-                   ' fetch or commit request for a group that it is not a'
-                   ' coordinator for.')
+    message = 'NOT_COORDINATOR'
+    description = ('The broker returns this error code if it is not the correct'
+                   ' coordinator for the specified consumer or transaction group')
     retriable = True
 
 
diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py
index 24b84a9..96a50cb 100644
--- a/kafka/producer/sender.py
+++ b/kafka/producer/sender.py
@@ -271,7 +271,7 @@ class Sender(threading.Thread):
             for batch in batches:
                 self._complete_batch(batch, None, -1)
 
-    def _fail_batch(batch, *args, **kwargs):
+    def _fail_batch(self, batch, *args, **kwargs):
         if self._transaction_state and self._transaction_state.producer_id_and_epoch.producer_id == batch.producer_id:
             # Reset the transaction state since we have hit an irrecoverable exception and cannot make any guarantees
             # about the previously committed message. Note that this will discard the producer id and sequence
diff --git a/kafka/protocol/fetch.py b/kafka/protocol/fetch.py
index d193eaf..036a37e 100644
--- a/kafka/protocol/fetch.py
+++ b/kafka/protocol/fetch.py
@@ -1,9 +1,15 @@
 from __future__ import absolute_import
 
+import collections
+
 from kafka.protocol.api import Request, Response
 from kafka.protocol.types import Array, Int8, Int16, Int32, Int64, Schema, String, Bytes
 
 
+AbortedTransaction = collections.namedtuple("AbortedTransaction",
+    ["producer_id", "first_offset"])
+
+
 class FetchResponse_v0(Response):
     API_KEY = 1
     API_VERSION = 0
diff --git a/kafka/record/abc.py b/kafka/record/abc.py
index df71785..c78f0da 100644
--- a/kafka/record/abc.py
+++ b/kafka/record/abc.py
@@ -110,6 +110,16 @@ class ABCRecordBatch(object):
             if needed.
         """
 
+    @abc.abstractproperty
+    def base_offset(self):
+        """ Return base offset for batch
+        """
+
+    @abc.abstractproperty
+    def size_in_bytes(self):
+        """ Return size of batch in bytes (includes header overhead)
+        """
+
     @abc.abstractproperty
     def magic(self):
         """ Return magic value (0, 1, 2) for batch.
diff --git a/kafka/record/default_records.py b/kafka/record/default_records.py
index 855306b..2158c48 100644
--- a/kafka/record/default_records.py
+++ b/kafka/record/default_records.py
@@ -104,6 +104,9 @@ class DefaultRecordBase(object):
 
     LOG_APPEND_TIME = 1
     CREATE_TIME = 0
+    NO_PRODUCER_ID = -1
+    NO_SEQUENCE = -1
+    MAX_INT = 2147483647
 
     def _assert_has_codec(self, compression_type):
         if compression_type == self.CODEC_GZIP:
@@ -136,6 +139,10 @@ class DefaultRecordBatch(DefaultRecordBase, ABCRecordBatch):
     def base_offset(self):
         return self._header_data[0]
 
+    @property
+    def size_in_bytes(self):
+        return self._header_data[1] + self.AFTER_LEN_OFFSET
+
     @property
     def leader_epoch(self):
         return self._header_data[2]
@@ -156,6 +163,14 @@ class DefaultRecordBatch(DefaultRecordBase, ABCRecordBatch):
     def last_offset_delta(self):
         return self._header_data[6]
 
+    @property
+    def last_offset(self):
+        return self.base_offset + self.last_offset_delta
+
+    @property
+    def next_offset(self):
+        return self.last_offset + 1
+
     @property
     def compression_type(self):
         return self.attributes & self.CODEC_MASK
@@ -180,6 +195,36 @@ class DefaultRecordBatch(DefaultRecordBase, ABCRecordBatch):
     def max_timestamp(self):
         return self._header_data[8]
 
+    @property
+    def producer_id(self):
+        return self._header_data[9]
+
+    def has_producer_id(self):
+        return self.producer_id > self.NO_PRODUCER_ID
+
+    @property
+    def producer_epoch(self):
+        return self._header_data[10]
+
+    @property
+    def base_sequence(self):
+        return self._header_data[11]
+
+    @property
+    def last_sequence(self):
+        if self.base_sequence == self.NO_SEQUENCE:
+            return self.NO_SEQUENCE
+        return self._increment_sequence(self.base_sequence, self.last_offset_delta)
+
+    def _increment_sequence(self, base, increment):
+        if base > (self.MAX_INT - increment):
+            return increment - (self.MAX_INT - base) - 1
+        return base + increment
+
+    @property
+    def records_count(self):
+        return self._header_data[12]
+
     def _maybe_uncompress(self):
         if not self._decompressed:
             compression_type = self.compression_type
diff --git a/kafka/record/legacy_records.py b/kafka/record/legacy_records.py
index 920b4fc..c126374 100644
--- a/kafka/record/legacy_records.py
+++ b/kafka/record/legacy_records.py
@@ -129,7 +129,7 @@ class LegacyRecordBase(object):
 
 class LegacyRecordBatch(ABCRecordBatch, LegacyRecordBase):
 
-    __slots__ = ("_buffer", "_magic", "_offset", "_crc", "_timestamp",
+    __slots__ = ("_buffer", "_magic", "_offset", "_length", "_crc", "_timestamp",
                  "_attributes", "_decompressed")
 
     def __init__(self, buffer, magic):
@@ -141,11 +141,20 @@ class LegacyRecordBatch(ABCRecordBatch, LegacyRecordBase):
         assert magic == magic_
 
         self._offset = offset
+        self._length = length
         self._crc = crc
         self._timestamp = timestamp
         self._attributes = attrs
         self._decompressed = False
 
+    @property
+    def base_offset(self):
+        return self._offset
+
+    @property
+    def size_in_bytes(self):
+        return self._length + self.LOG_OVERHEAD
+
     @property
     def timestamp_type(self):
         """0 for CreateTime; 1 for LogAppendTime; None if unsupported.


Test patch:
diff --git a/test/test_admin_integration.py b/test/test_admin_integration.py
index 83b6cca..f95f367 100644
--- a/test/test_admin_integration.py
+++ b/test/test_admin_integration.py
@@ -9,7 +9,7 @@ from time import time, sleep
 from kafka.admin import (
     ACLFilter, ACLOperation, ACLPermissionType, ResourcePattern, ResourceType, ACL, ConfigResource, ConfigResourceType)
 from kafka.errors import (
-        BrokerResponseError, KafkaError, NoError, GroupCoordinatorNotAvailableError, NonEmptyGroupError, 
+        BrokerResponseError, KafkaError, NoError, CoordinatorNotAvailableError, NonEmptyGroupError,
         GroupIdNotFoundError, OffsetOutOfRangeError, UnknownTopicOrPartitionError)
 
 
@@ -150,7 +150,7 @@ def test_describe_configs_invalid_broker_id_raises(kafka_admin_client):
 def test_describe_consumer_group_does_not_exist(kafka_admin_client):
     """Tests that the describe consumer group call fails if the group coordinator is not available
     """
-    with pytest.raises(GroupCoordinatorNotAvailableError):
+    with pytest.raises(CoordinatorNotAvailableError):
         kafka_admin_client.describe_consumer_groups(['test'])
 
 
diff --git a/test/test_coordinator.py b/test/test_coordinator.py
index 1d1a6df..00a9293 100644
--- a/test/test_coordinator.py
+++ b/test/test_coordinator.py
@@ -444,7 +444,7 @@ def test_send_offset_commit_request_fail(mocker, patched_coord, offsets):
     # No coordinator
     ret = patched_coord._send_offset_commit_request(offsets)
     assert ret.failed()
-    assert isinstance(ret.exception, Errors.GroupCoordinatorNotAvailableError)
+    assert isinstance(ret.exception, Errors.CoordinatorNotAvailableError)
 
 
 @pytest.mark.parametrize('api_version,req_type', [
@@ -497,11 +497,11 @@ def test_send_offset_commit_request_success(mocker, patched_coord, offsets):
     (OffsetCommitResponse[0]([('foobar', [(0, 28), (1, 28)])]),
      Errors.InvalidCommitOffsetSizeError, False),
     (OffsetCommitResponse[0]([('foobar', [(0, 14), (1, 14)])]),
-     Errors.GroupLoadInProgressError, False),
+     Errors.CoordinatorLoadInProgressError, False),
     (OffsetCommitResponse[0]([('foobar', [(0, 15), (1, 15)])]),
-     Errors.GroupCoordinatorNotAvailableError, True),
+     Errors.CoordinatorNotAvailableError, True),
     (OffsetCommitResponse[0]([('foobar', [(0, 16), (1, 16)])]),
-     Errors.NotCoordinatorForGroupError, True),
+     Errors.NotCoordinatorError, True),
     (OffsetCommitResponse[0]([('foobar', [(0, 7), (1, 7)])]),
      Errors.RequestTimedOutError, True),
     (OffsetCommitResponse[0]([('foobar', [(0, 25), (1, 25)])]),
@@ -557,7 +557,7 @@ def test_send_offset_fetch_request_fail(mocker, patched_coord, partitions):
     # No coordinator
     ret = patched_coord._send_offset_fetch_request(partitions)
     assert ret.failed()
-    assert isinstance(ret.exception, Errors.GroupCoordinatorNotAvailableError)
+    assert isinstance(ret.exception, Errors.CoordinatorNotAvailableError)
 
 
 @pytest.mark.parametrize('api_version,req_type', [
@@ -606,9 +606,9 @@ def test_send_offset_fetch_request_success(patched_coord, partitions):
 
 @pytest.mark.parametrize('response,error,dead', [
     (OffsetFetchResponse[0]([('foobar', [(0, 123, '', 14), (1, 234, '', 14)])]),
-     Errors.GroupLoadInProgressError, False),
+     Errors.CoordinatorLoadInProgressError, False),
     (OffsetFetchResponse[0]([('foobar', [(0, 123, '', 16), (1, 234, '', 16)])]),
-     Errors.NotCoordinatorForGroupError, True),
+     Errors.NotCoordinatorError, True),
     (OffsetFetchResponse[0]([('foobar', [(0, 123, '', 25), (1, 234, '', 25)])]),
      Errors.UnknownMemberIdError, False),
     (OffsetFetchResponse[0]([('foobar', [(0, 123, '', 22), (1, 234, '', 22)])]),
diff --git a/test/test_fetcher.py b/test/test_fetcher.py
index 184acc9..3fc0c55 100644
--- a/test/test_fetcher.py
+++ b/test/test_fetcher.py
@@ -452,7 +452,7 @@ def test__unpack_records(mocker):
         (None, b"c", None),
     ]
     memory_records = MemoryRecords(_build_record_batch(messages))
-    part_records = Fetcher.PartitionRecords(0, tp, memory_records, None, None, False, mocker.MagicMock(), lambda x: None)
+    part_records = Fetcher.PartitionRecords(0, tp, memory_records)
     records = list(part_records.record_iterator)
     assert len(records) == 3
     assert all(map(lambda x: isinstance(x, ConsumerRecord), records))
@@ -557,7 +557,7 @@ def test_partition_records_offset(mocker):
     tp = TopicPartition('foo', 0)
     messages = [(None, b'msg', None) for i in range(batch_start, batch_end)]
     memory_records = MemoryRecords(_build_record_batch(messages, offset=batch_start))
-    records = Fetcher.PartitionRecords(fetch_offset, tp, memory_records, None, None, False, mocker.MagicMock(), lambda x: None)
+    records = Fetcher.PartitionRecords(fetch_offset, tp, memory_records)
     assert records
     assert records.next_fetch_offset == fetch_offset
     msgs = records.take(1)
@@ -574,7 +574,7 @@ def test_partition_records_offset(mocker):
 def test_partition_records_empty(mocker):
     tp = TopicPartition('foo', 0)
     memory_records = MemoryRecords(_build_record_batch([]))
-    records = Fetcher.PartitionRecords(0, tp, memory_records, None, None, False, mocker.MagicMock(), lambda x: None)
+    records = Fetcher.PartitionRecords(0, tp, memory_records)
     msgs = records.take()
     assert len(msgs) == 0
     assert not records
@@ -587,7 +587,7 @@ def test_partition_records_no_fetch_offset(mocker):
     tp = TopicPartition('foo', 0)
     messages = [(None, b'msg', None) for i in range(batch_start, batch_end)]
     memory_records = MemoryRecords(_build_record_batch(messages, offset=batch_start))
-    records = Fetcher.PartitionRecords(fetch_offset, tp, memory_records, None, None, False, mocker.MagicMock(), lambda x: None)
+    records = Fetcher.PartitionRecords(fetch_offset, tp, memory_records)
     msgs = records.take()
     assert len(msgs) == 0
     assert not records
@@ -611,7 +611,7 @@ def test_partition_records_compacted_offset(mocker):
             builder.append(key=None, value=b'msg', timestamp=None, headers=[])
     builder.close()
     memory_records = MemoryRecords(builder.buffer())
-    records = Fetcher.PartitionRecords(fetch_offset, tp, memory_records, None, None, False, mocker.MagicMock(), lambda x: None)
+    records = Fetcher.PartitionRecords(fetch_offset, tp, memory_records)
     msgs = records.take()
     assert len(msgs) == batch_end - fetch_offset - 1
     assert msgs[0].offset == fetch_offset + 1