[KIP-932] : Implement Share consumer interface with poll API

.semaphore/semaphore.yml

@@ -9,6 +9,9 @@ global_job_config:
   env_vars:
     - name: LIBRDKAFKA_VERSION
       value: v2.14.0
+    # TODO KIP-932: Remove LIBRDKAFKA_BRANCH once LIBRDKAFKA_VERSION includes share consumer support
+    - name: LIBRDKAFKA_BRANCH
+      value: dev_kip-932_queues-for-kafka
   prologue:
     commands:
       - checkout
@@ -32,6 +35,7 @@ blocks:
             - export ARCH=x64
             - sem-version python 3.11
             - pip install uv
+            # TODO KIP-932: Add LIBRDKAFKA_BRANCH fallback for share consumer support
             - PIP_INSTALL_OPTIONS="--user" tools/wheels/build-wheels.sh "${LIBRDKAFKA_VERSION#v}" wheelhouse 2.16.2
             - tar -czf wheelhouse-macOS-${ARCH}.tgz wheelhouse
             - artifact push workflow wheelhouse-macOS-${ARCH}.tgz --destination artifacts/wheels-${OS_NAME}-${ARCH}.tgz/
@@ -57,6 +61,7 @@ blocks:
             - export ARCH=x64
             - sem-version python 3.13
             - pip install uv
+            # TODO KIP-932: Add LIBRDKAFKA_BRANCH fallback for share consumer support
             - PIP_INSTALL_OPTIONS="--user" tools/wheels/build-wheels.sh "${LIBRDKAFKA_VERSION#v}" wheelhouse
             - tar -czf wheelhouse-macOS-${ARCH}-py313-plus.tgz wheelhouse
             - artifact push workflow wheelhouse-macOS-${ARCH}-py313-plus.tgz --destination artifacts/wheels-${OS_NAME}-${ARCH}-py313-plus.tgz/
@@ -80,6 +85,7 @@ blocks:
             - export ARCH=arm64
             - sem-version python 3.11
             - pip install uv
+            # TODO KIP-932: Add LIBRDKAFKA_BRANCH fallback for share consumer support
             - PIP_INSTALL_OPTIONS="--user" tools/wheels/build-wheels.sh "${LIBRDKAFKA_VERSION#v}" wheelhouse 2.16.2
             - tar -czf wheelhouse-macOS-${ARCH}.tgz wheelhouse
             - artifact push workflow wheelhouse-macOS-${ARCH}.tgz --destination artifacts/wheels-${OS_NAME}-${ARCH}.tgz/
@@ -107,6 +113,7 @@ blocks:
             - export ARCH=arm64
             - sem-version python 3.13
             - pip install uv
+            # TODO KIP-932: Add LIBRDKAFKA_BRANCH fallback for share consumer support
             - PIP_INSTALL_OPTIONS="--user" tools/wheels/build-wheels.sh "${LIBRDKAFKA_VERSION#v}" wheelhouse
             - tar -czf wheelhouse-macOS-${ARCH}-py313-plus.tgz wheelhouse
             - artifact push workflow wheelhouse-macOS-${ARCH}-py313-plus.tgz --destination artifacts/wheels-${OS_NAME}-${ARCH}-py313-plus.tgz/
@@ -128,6 +135,7 @@ blocks:
             - export ARCH=arm64
             - sem-version python 3.13
             - pip install uv
+            # TODO KIP-932: Add LIBRDKAFKA_BRANCH fallback for share consumer support
             - ./tools/wheels/build-wheels.sh "${LIBRDKAFKA_VERSION#v}" wheelhouse
             - tar -czf wheelhouse-linux-${ARCH}.tgz wheelhouse
             - artifact push workflow wheelhouse-linux-${ARCH}.tgz --destination artifacts/wheels-${OS_NAME}-${ARCH}.tgz/
@@ -149,6 +157,7 @@ blocks:
             - export ARCH=x64
             - sem-version python 3.11
             - pip install uv
+            # TODO KIP-932: Add LIBRDKAFKA_BRANCH fallback for share consumer support
             - ./tools/wheels/build-wheels.sh "${LIBRDKAFKA_VERSION#v}" wheelhouse
             - tar -czf wheelhouse-linux-${ARCH}.tgz wheelhouse
             - artifact push workflow wheelhouse-linux-${ARCH}.tgz --destination artifacts/wheels-${OS_NAME}-${ARCH}.tgz/
@@ -181,6 +190,7 @@ blocks:
           commands:
             - sem-version python 3.11.9
             - bash tools/mingw-w64/semaphore_commands.sh
+            # TODO KIP-932: Add LIBRDKAFKA_BRANCH fallback for share consumer support
             - bash tools/wheels/install-librdkafka.sh $env:LIBRDKAFKA_VERSION.TrimStart("v") dest
             - tools/wheels/build-wheels.bat x64 win_amd64 dest wheelhouse
             - tar -czf wheelhouse-windows-${Env:ARCH}.tgz wheelhouse
@@ -319,9 +329,15 @@ blocks:
             # Install existing test requirements
             - uv pip install -r requirements/requirements-tests-install.txt
 
+            # TODO KIP-932: Remove LIBRDKAFKA_BRANCH fallback once LIBRDKAFKA_VERSION includes share consumer support
             # Build and install confluent-kafka from source
             - lib_dir=dest/runtimes/$OS_NAME-$ARCH/native
-            - tools/wheels/install-librdkafka.sh "${LIBRDKAFKA_VERSION#v}" dest
+            - |
+              if [[ -n $LIBRDKAFKA_BRANCH ]]; then
+                tools/wheels/build-librdkafka-branch.sh "$LIBRDKAFKA_BRANCH" dest
+              else
+                tools/wheels/install-librdkafka.sh "${LIBRDKAFKA_VERSION#v}" dest
+              fi
             - export CFLAGS="$CFLAGS -I${PWD}/dest/build/native/include"
             - export LDFLAGS="$LDFLAGS -L${PWD}/${lib_dir}"
             - export LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$PWD/$lib_dir"

setup.py

@@ -23,6 +23,7 @@
         os.path.join(ext_dir, 'confluent_kafka.c'),
         os.path.join(ext_dir, 'Producer.c'),
         os.path.join(ext_dir, 'Consumer.c'),
+        os.path.join(ext_dir, 'ShareConsumer.c'),
         os.path.join(ext_dir, 'Metadata.c'),
         os.path.join(ext_dir, 'AdminTypes.c'),
         os.path.join(ext_dir, 'Admin.c'),

src/confluent_kafka/__init__.py

@@ -39,6 +39,7 @@
     Consumer,
     Message,
     Producer,
+    ShareConsumer,
     TopicPartition,
     Uuid,
     consistent,
@@ -54,6 +55,7 @@
 __all__ = [
     "admin",
     "Consumer",
+    "ShareConsumer",
     "aio",
     "KafkaError",
     "KafkaException",

src/confluent_kafka/cimpl.pyi

@@ -63,12 +63,14 @@ class KafkaError:
     DELEGATION_TOKEN_REQUEST_NOT_ALLOWED: int
     DUPLICATE_RESOURCE: int
     DUPLICATE_SEQUENCE_NUMBER: int
+    DUPLICATE_VOTER: int
     ELECTION_NOT_NEEDED: int
     ELIGIBLE_LEADERS_NOT_AVAILABLE: int
     FEATURE_UPDATE_FAILED: int
     FENCED_INSTANCE_ID: int
     FENCED_LEADER_EPOCH: int
     FENCED_MEMBER_EPOCH: int
+    FENCED_STATE_EPOCH: int
     FETCH_SESSION_ID_NOT_FOUND: int
     GROUP_AUTHORIZATION_FAILED: int
     GROUP_ID_NOT_FOUND: int
@@ -89,20 +91,26 @@ class KafkaError:
     INVALID_PRODUCER_EPOCH: int
     INVALID_PRODUCER_ID_MAPPING: int
     INVALID_RECORD: int
+    INVALID_RECORD_STATE: int
+    INVALID_REGISTRATION: int
+    INVALID_REGULAR_EXPRESSION: int
     INVALID_REPLICATION_FACTOR: int
     INVALID_REPLICA_ASSIGNMENT: int
     INVALID_REQUEST: int
     INVALID_REQUIRED_ACKS: int
+    INVALID_SHARE_SESSION_EPOCH: int
     INVALID_SESSION_TIMEOUT: int
     INVALID_TIMESTAMP: int
     INVALID_TRANSACTION_TIMEOUT: int
     INVALID_TXN_STATE: int
     INVALID_UPDATE_VERSION: int
+    INVALID_VOTER_KEY: int
     KAFKA_STORAGE_ERROR: int
     LEADER_NOT_AVAILABLE: int
     LISTENER_NOT_FOUND: int
     LOG_DIR_NOT_FOUND: int
     MEMBER_ID_REQUIRED: int
+    MISMATCHED_ENDPOINT_TYPE: int
     MSG_SIZE_TOO_LARGE: int
     NETWORK_EXCEPTION: int
     NON_EMPTY_GROUP: int
@@ -131,19 +139,26 @@ class KafkaError:
     RESOURCE_NOT_FOUND: int
     SASL_AUTHENTICATION_FAILED: int
     SECURITY_DISABLED: int
+    SHARE_SESSION_LIMIT_REACHED: int
+    SHARE_SESSION_NOT_FOUND: int
     STALE_BROKER_EPOCH: int
     STALE_CTRL_EPOCH: int
     STALE_MEMBER_EPOCH: int
+    STREAMS_INVALID_TOPOLOGY: int
+    STREAMS_INVALID_TOPOLOGY_EPOCH: int
+    STREAMS_TOPOLOGY_FENCED: int
     TELEMETRY_TOO_LARGE: int
     THROTTLING_QUOTA_EXCEEDED: int
     TOPIC_ALREADY_EXISTS: int
     TOPIC_AUTHORIZATION_FAILED: int
     TOPIC_DELETION_DISABLED: int
     TOPIC_EXCEPTION: int
+    TRANSACTION_ABORTABLE: int
     TRANSACTIONAL_ID_AUTHORIZATION_FAILED: int
     TRANSACTION_COORDINATOR_FENCED: int
     UNACCEPTABLE_CREDENTIAL: int
     UNKNOWN: int
+    UNKNOWN_CONTROLLER_ID: int
     UNKNOWN_LEADER_EPOCH: int
     UNKNOWN_MEMBER_ID: int
     UNKNOWN_PRODUCER_ID: int
@@ -154,9 +169,11 @@ class KafkaError:
     UNSTABLE_OFFSET_COMMIT: int
     UNSUPPORTED_ASSIGNOR: int
     UNSUPPORTED_COMPRESSION_TYPE: int
+    UNSUPPORTED_ENDPOINT_TYPE: int
     UNSUPPORTED_FOR_MESSAGE_FORMAT: int
     UNSUPPORTED_SASL_MECHANISM: int
     UNSUPPORTED_VERSION: int
+    VOTER_NOT_FOUND: int
     _ALL_BROKERS_DOWN: int
     _APPLICATION: int
     _ASSIGNMENT_LOST: int
@@ -535,6 +552,22 @@ class Consumer:
     def memberid(self) -> str: ...
     def set_sasl_credentials(self, username: str, password: str) -> None: ...
 
+class ShareConsumer:
+    """Share Consumer for queue-like message consumption (KIP-932)."""
+    @overload
+    def __init__(self, config: Dict[str, Any]) -> None: ...
+    @overload
+    def __init__(self, config: Dict[str, Any], /, **kwargs: Any) -> None: ...
+    @overload
+    def __init__(self, **config: Any) -> None: ...
+    def subscribe(self, topics: List[str]) -> None: ...
+    def unsubscribe(self) -> None: ...
+    def subscription(self) -> List[str]: ...
+    def poll(self, timeout: float = -1) -> List[Message]: ...
+    def close(self) -> None: ...
+    def __enter__(self) -> "ShareConsumer": ...
+    def __exit__(self, exc_type: Any, exc_value: Any, exc_traceback: Any) -> Optional[bool]: ...
+
 class _AdminClientImpl:
     def __init__(self, config: Dict[str, Union[str, int, float, bool]]) -> None: ...
     def __enter__(self) -> Self: ...