Add kafka.protocol.new.consumer.offsets w/ extra consts

kafka/protocol/new/consumer/__init__.py

@@ -1,9 +1,11 @@
 from .fetch import *
 from .group import *
+from .offsets import *
 
 
 __all__ = [
     'FetchRequest', 'FetchResponse',
+    'UNKNOWN_OFFSET', 'OffsetResetStrategy',
     'ListOffsetsRequest', 'ListOffsetsResponse',
     'JoinGroupRequest', 'JoinGroupResponse',
     'SyncGroupRequest', 'SyncGroupResponse',

kafka/protocol/new/consumer/fetch.py

@@ -4,15 +4,6 @@
 class FetchRequest(ApiMessage): pass
 class FetchResponse(ApiMessage): pass
 
-class ListOffsetsRequest(ApiMessage): pass
-class ListOffsetsResponse(ApiMessage): pass
-
-class OffsetForLeaderEpochRequest(ApiMessage): pass
-class OffsetForLeaderEpochResponse(ApiMessage): pass
-
-
 __all__ = [
     'FetchRequest', 'FetchResponse',
-    'ListOffsetsRequest', 'ListOffsetsResponse',
-    'OffsetForLeaderEpochRequest', 'OffsetForLeaderEpochResponse',
 ]

kafka/protocol/new/consumer/offsets.py

@@ -0,0 +1,22 @@
+from ..api_message import ApiMessage
+
+
+UNKNOWN_OFFSET = -1
+
+class OffsetResetStrategy:
+    LATEST = -1
+    EARLIEST = -2
+    NONE = 0
+
+class ListOffsetsRequest(ApiMessage): pass
+class ListOffsetsResponse(ApiMessage): pass
+
+class OffsetForLeaderEpochRequest(ApiMessage): pass
+class OffsetForLeaderEpochResponse(ApiMessage): pass
+
+
+__all__ = [
+    'UNKNOWN_OFFSET', 'OffsetResetStrategy',
+    'ListOffsetsRequest', 'ListOffsetsResponse',
+    'OffsetForLeaderEpochRequest', 'OffsetForLeaderEpochResponse',
+]

test/protocol/new/consumer/test_new_list_offsets.py

@@ -1,6 +1,9 @@
 import pytest
 
-from kafka.protocol.new.consumer import ListOffsetsRequest, ListOffsetsResponse
+from kafka.protocol.new.consumer import (
+    ListOffsetsRequest, ListOffsetsResponse,
+    UNKNOWN_OFFSET, OffsetResetStrategy,
+)
 
 
 @pytest.mark.parametrize("version", range(ListOffsetsRequest.min_version, ListOffsetsRequest.max_version + 1))
@@ -54,3 +57,14 @@ def test_list_offsets_response_roundtrip(version):
     encoded = ListOffsetsResponse.encode(data, version=version)
     decoded = ListOffsetsResponse.decode(encoded, version=version)
     assert decoded == data
+
+
+def test_unknown_offset():
+    assert UNKNOWN_OFFSET == -1
+
+
+def test_offset_reset_strategies():
+    assert OffsetResetStrategy.LATEST == -1
+    assert OffsetResetStrategy.EARLIEST == -2
+    assert OffsetResetStrategy.NONE == 0
+