Set up type hinting: add missing types to Python files and C extension stubs #2041

fangnx · 2025-09-05T15:56:28Z

What

First PR to set up type hinting for this repo, according to https://peps.python.org/pep-0484/. This PR includes:

C extension stubs (cimpl.pyi). Note that stubgen didn't work well with our confluent_kafka module and only generated the stubs partially, so some manual parsing of C code (with AI) was used to fill in the gaps
Add types to Python files in the repository
Verify the types with mypy type checker: most files are validated successfully except for schema-registry module which is already typed prior to this PR. I will investigate in the follow-up PR and see if those are real issues or false positives

Checklist

Contains customer facing changes? Including API/behavior changes
Did you add sufficient unit test and/or integration test coverage for this PR?
- If not, please explain why it is not required

References

JIRA: https://confluentinc.atlassian.net/browse/DGS-22076

Test & Review

Open questions / Follow-ups

confluent-cla-assistant · 2025-09-05T15:56:39Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

Copilot

Pull Request Overview

This PR introduces initial type hinting support for the confluent_kafka Python package by adding type stubs for the C extension and typing annotations to core serialization classes.

Establishes type safety foundation with C extension stubs and centralized type definitions
Adds comprehensive type annotations to producer and consumer serialization classes
Creates type-aware error handling hierarchy

Reviewed Changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
src/confluent_kafka/cimpl.pyi	Type stubs for C extension classes and functions
src/confluent_kafka/_types.py	Centralized type aliases for the package
src/confluent_kafka/py.typed	Marker file indicating package supports typing
src/confluent_kafka/serializing_producer.py	Type annotations for SerializingProducer class
src/confluent_kafka/deserializing_consumer.py	Type annotations for DeserializingConsumer class
src/confluent_kafka/error.py	Type annotations for error classes
src/confluent_kafka/init.py	Type annotations for utility classes

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

Copilot · 2025-09-05T20:01:19Z

src/confluent_kafka/deserializing_consumer.py

        super(DeserializingConsumer, self).__init__(conf_copy)

-    def poll(self, timeout=-1):
+    def poll(self, timeout: float = -1) -> Optional['Message']:


The return type annotation uses forward reference quotes around 'Message' but Message is already imported on line 21. Remove the quotes to use the directly imported type.

Suggested change

def poll(self, timeout: float = -1) -> Optional['Message']:

def poll(self, timeout: float = -1) -> Optional[Message]:

MSeal

Would like to spend some time over zoom talking through the work on this PR but overall looks reasonable as a solution PR

MSeal · 2025-09-06T00:10:33Z

src/confluent_kafka/_model/__init__.py

    """

-    def __init__(self, group_id, topic_partitions=None):
+    def __init__(self, group_id: str, topic_partitions: Optional[List['cimpl.TopicPartition']] = None) -> None:


Why'd this need the cimpl path prefix to work here? Curious to chat about how the cbinding type errors were worked out in development

We are only importing the module from .. import cimpl so the type checker would not find the directly referenced TopicPartition. We can switch to from ..cimpl import TopicPartition to make the code a bit cleaner

MSeal · 2025-09-06T00:12:12Z

src/confluent_kafka/_util/conversion_util.py

 from enum import Enum

+# Generic type for enum conversion
+E = TypeVar('E', bound=Enum)


Why do we need this? Can't we just use type[Enum] below?

Yes I think we can just leave it type[Enum]. I think Type[E] creates a stricter type relationship with the specific enum class, but it overcomplicates the code a bit and I don't think this function (meant to handle types itself) can benefit from it

MSeal · 2025-09-06T00:13:11Z

src/confluent_kafka/admin/_cluster.py

        self.controller = controller
        self.nodes = nodes
-        self.authorized_operations = None
+        self.authorized_operations: Optional[List[AclOperation]] = None


Why the hint here and why does it differ from the hint in the def?

Fixed. Also update authorized_operations to be more specific: Optional[List[Union[str, int, AclOperation]]], as those 3 types are handled in ConversionUtil.convert_to_enum

src/confluent_kafka/admin/_scram.py

MSeal · 2025-09-06T00:22:13Z

src/confluent_kafka/cimpl.pyi

@@ -0,0 +1,443 @@
+"""


I do dislike how we'll have two locations for type/actual definitions we need to maintain now. However it may be the ideal solution for now. This really makes me want to try a pass on converting the c files to Cython and then getting the type hinting for free while avoiding manually defining so many Python things in C binding calls. Not in scope for this PR though it would eliminate the above dual definition problem. For now let's add a warning comment to the C code files to update this file when/if editing interfaces therein

… entrypoint init files

Copilot

Pull Request Overview

Copilot reviewed 43 out of 44 changed files in this pull request and generated 6 comments.

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

Copilot · 2025-10-20T22:17:19Z

src/confluent_kafka/experimental/aio/_common.py

-    def log(self, *args, **kwargs):
-        self.loop.call_soon_threadsafe(self.logger.log, *args, **kwargs)
+    def log(self, *args: Any, **kwargs: Any) -> None:
+        self.loop.call_soon_threadsafe(lambda: self.logger.log(*args, **kwargs))


The original implementation used self.loop.call_soon_threadsafe(self.logger.log, *args, **kwargs) which directly passes the method and arguments to be executed in the event loop. The new implementation wraps this in a lambda, which means the arguments are captured immediately (when log() is called) rather than being evaluated in the event loop thread. This could lead to subtle bugs if the arguments are mutable and modified between the time log() is called and when the lambda executes. Revert to the original implementation: self.loop.call_soon_threadsafe(self.logger.log, *args, **kwargs)

Suggested change

self.loop.call_soon_threadsafe(lambda: self.logger.log(*args, **kwargs))

self.loop.call_soon_threadsafe(self.logger.log, *args, **kwargs)

src/confluent_kafka/admin/_config.py

src/confluent_kafka/admin/_group.py

Copilot · 2025-10-20T22:17:20Z

src/confluent_kafka/admin/_listoffsets.py

-        self.leader_epoch = leader_epoch
-        if self.leader_epoch < 0:
-            self.leader_epoch = None
+        self.leader_epoch: Optional[int] = leader_epoch if leader_epoch >= 0 else None


[nitpick] The conditional logic is now inline in the assignment which changes behavior from the original implementation. The original code set self.leader_epoch = leader_epoch first and then conditionally set it to None. This new implementation doesn't preserve the original value semantics if leader_epoch is negative. Consider keeping closer to the original pattern for clarity.

Suggested change

self.leader_epoch: Optional[int] = leader_epoch if leader_epoch >= 0 else None

self.leader_epoch: Optional[int] = leader_epoch

if leader_epoch < 0:

self.leader_epoch = None

Copilot · 2025-10-20T22:17:21Z

src/confluent_kafka/admin/__init__.py

                                              AdminClient._make_consumer_groups_result)

-        super(AdminClient, self).describe_consumer_groups(group_ids, f, **kwargs)
+        super(AdminClient, self).describe_consumer_groups(group_ids, f, **kwargs)  # type: ignore[arg-type]


The type: ignore[arg-type] comment suggests a type mismatch between the parent class signature and the override. This indicates the stub file (cimpl.pyi) may have incorrect type hints for _AdminClientImpl.describe_consumer_groups. Review and fix the stub file types instead of suppressing the error.

Copilot · 2025-10-20T22:17:21Z

src/confluent_kafka/deserializing_consumer.py

                key = self._key_deserializer(key, ctx)
            except Exception as se:
                raise KeyDeserializationError(exception=se, kafka_message=msg)



The type: ignore comments indicate that set_key and set_value in the stub file expect bytes but the deserialized key/value can be of any type. The stub file signature for these methods should be updated to accept Any instead of just bytes, or these methods should be removed from the public API if they're intended to be internal-only.

Suggested change

# The stub file for Message.set_key/set_value expects bytes, but after deserialization, these can be any type.

# The stub should be updated to accept Any. See CodeQL rule and issue for details.

Copilot

Pull Request Overview

Copilot reviewed 43 out of 44 changed files in this pull request and generated 2 comments.

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

Copilot · 2025-10-21T18:00:58Z

src/confluent_kafka/admin/_group.py

-        self.topic_partitions = topic_partitions
-        if self.topic_partitions is None:
-            self.topic_partitions = []
+    def __init__(self, topic_partitions: List[TopicPartition] = []) -> None:


Mutable default argument [] for topic_partitions parameter. This creates a shared list across all instances. Use None as default and handle it in the method body (which is already done on line 84, so just change default to None).

Suggested change

def __init__(self, topic_partitions: List[TopicPartition] = []) -> None:

def __init__(self, topic_partitions: Optional[List[TopicPartition]] = None) -> None:

This is catching a real bug here -- we should fix this to be None as it's currently sharing the instance value incorrectly across instances without provided inputs

Copilot · 2025-10-21T18:00:59Z

src/confluent_kafka/admin/_listoffsets.py

+        self.leader_epoch: Optional[int] = leader_epoch
+        if leader_epoch < 0:
            self.leader_epoch = None


The variable self.leader_epoch is assigned on line 162, then immediately conditionally modified on line 164. This could be simplified by directly assigning the conditional result: self.leader_epoch = None if leader_epoch < 0 else leader_epoch.

Suggested change

self.leader_epoch: Optional[int] = leader_epoch

if leader_epoch < 0:

self.leader_epoch = None

self.leader_epoch: Optional[int] = None if leader_epoch < 0 else leader_epoch

MSeal

Some minor changes requested, then good to go

MSeal · 2025-10-21T21:16:36Z

src/confluent_kafka/_model/__init__.py

    READ_COMMITTED = cimpl.ISOLATION_LEVEL_READ_COMMITTED  #: Skip offsets belonging to an aborted transaction.

-    def __lt__(self, other):
+    def __lt__(self, other) -> Any:


should be -> bool

MSeal · 2025-10-21T21:16:41Z

src/confluent_kafka/_model/__init__.py

    UNCLEAN = cimpl.ELECTION_TYPE_UNCLEAN

-    def __lt__(self, other):
+    def __lt__(self, other) -> Any:


should be -> bool

MSeal · 2025-10-21T21:42:18Z

src/confluent_kafka/admin/__init__.py

        return super(AdminClient, self).list_groups(*args, **kwargs)

-    def create_partitions(self, new_partitions, **kwargs):
+    def create_partitions(  # type: ignore[override]


Why do we need the type ignore here?

I think it's unfortunately required to work around the current setup:

In C extension layer, Admin_create_partitions returns None

In the cimpl.pyi stub file where we define the C-extension class _AdminClientImpl, we set return type of create_partitions to None

In the python wrapper layer, AdminClient inherits _AdminClientImpl but create_partitions returns a future instead of None

MSeal · 2025-10-21T21:43:49Z

src/confluent_kafka/admin/__init__.py


-    def create_partitions(self, new_partitions, **kwargs):
+    def create_partitions(  # type: ignore[override]
+        self, new_partitions: List[NewPartitions], **kwargs: Any


TIL **kwargs you only set the value type on the hint

Yep, I assume the reason is that the type of key is always str

MSeal · 2025-10-21T21:51:26Z

src/confluent_kafka/admin/_config.py

        return hash((self.restype, self.name))

-    def __lt__(self, other):
+    def __lt__(self, other: 'ConfigResource') -> bool:


Enforcement here might be awkward, since technically __lt__ should be handling non-type match argumentsand giving a default answer or raising is NotImplemented -- which it appears to not implement

Good point. I think maybe I should add the type check here and in other places to follow the pattern below:

def __lt__(self, other: object) -> bool: if not isinstance(other, MyClass): return NotImplemented return self.attribute < other.attribute

Yes I think it should be that

MSeal · 2025-10-21T21:55:00Z

src/confluent_kafka/admin/_listoffsets.py

            return cls.for_timestamp(index)

-    def __lt__(self, other):
+    def __lt__(self, other) -> Any:


other missing type

MSeal · 2025-10-21T21:55:59Z

src/confluent_kafka/admin/_metadata.py


-    def __init__(self):
+    def __init__(self) -> None:
        self.topic = None


Missing string type if we're also typing partitions

MSeal · 2025-10-21T21:56:20Z

src/confluent_kafka/admin/_metadata.py

-        self.partitions = {}
+        self.partitions: Dict[int, 'PartitionMetadata'] = {}
        """Map of partitions indexed by partition id. Value is a PartitionMetadata object."""
        self.error = None


Missing KafkaError type if we're also typing partitions

MSeal · 2025-10-21T21:56:59Z

src/confluent_kafka/admin/_scram.py

    SCRAM_SHA_512 = cimpl.SCRAM_MECHANISM_SHA_512  #: SCRAM-SHA-512 mechanism

-    def __lt__(self, other):
+    def __lt__(self, other) -> Any:


bool return type

MSeal · 2025-10-21T21:58:51Z

src/confluent_kafka/experimental/aio/producer/_AIOProducer.py

            # Update buffer activity since we just flushed
            self._buffer_timeout_manager.mark_activity()

-        # Then flush the underlying producer and wait for delivery confirmation


I think the old text was grammatically correct

Cursor was trying to smart and made a bunch of creative text edits :(

fangnx · 2025-10-22T19:33:55Z

src/confluent_kafka/experimental/aio/_common.py

-    def log(self, *args, **kwargs):
-        self.loop.call_soon_threadsafe(self.logger.log, *args, **kwargs)
+    def log(self, *args: Any, **kwargs: Any) -> None:
+        self.loop.call_soon_threadsafe(lambda: self.logger.log(*args, **kwargs))


Another fix to pass the function in correctly

def call_soon_threadsafe(self, callback, *args, context=None)

sonarqube-confluent · 2025-10-22T19:50:48Z

Analysis Details

16 Issues

0 Bugs
0 Vulnerabilities
16 Code Smells

Coverage and Duplications

85.90% Coverage (67.10% Estimated after merge)
No duplication information (4.80% Estimated after merge)

Project ID: confluent-kafka-python

View in SonarQube

MSeal · 2025-10-22T22:04:01Z

src/confluent_kafka/experimental/aio/producer/_kafka_batch_executor.py

-        self,
-        batch_messages: List[Dict[str, Any]]
-    ) -> None:
+    async def flush_librdkafka_queue(self, timeout=-1):


I think since you picked up master changes you now have some more methods to type -- I'll approve PR and we can do a second pass for these separately as part of putting code checks on typing.

fangnx marked this pull request as ready for review September 5, 2025 18:45

Copilot AI review requested due to automatic review settings September 5, 2025 18:45

fangnx requested review from a team and MSeal as code owners September 5, 2025 18:45

This comment was marked as outdated.

Sign in to view

fangnx changed the title ~~Add type hinting~~ Set up type hinting: C extension stubs + typing for serializing_producer + deserializing_consumer Sep 5, 2025

fangnx requested a review from Copilot September 5, 2025 20:00

Copilot AI reviewed Sep 5, 2025

View reviewed changes

This comment has been minimized.

Sign in to view

MSeal reviewed Sep 6, 2025

View reviewed changes

fangnx requested a review from MSeal September 9, 2025 21:21

fangnx added 8 commits October 14, 2025 23:29

update

ec73a83

remove py.typed for now

8e7f279

update

e55f45e

fix cimply and add types to serde producer/consumer

c875eeb

admin

04c744b

address feedback

57855d1

add warning to stub and c files; admin typing more

01060bf

add accidentally removed md files

c6b0192

airlock-confluentinc bot force-pushed the typehinting-kafka branch from 7b378e7 to c6b0192 Compare October 15, 2025 03:32

fix merge conflicts in md files, add types to admin and serialization…

5a022be

… entrypoint init files

This comment has been minimized.

Sign in to view

finish admin init

68c5816

This comment has been minimized.

Sign in to view

fangnx added 3 commits October 15, 2025 15:55

add types for AIO module

84f4ca0

add types for SR rules

15bfcba

linter fix

c28e856

fangnx changed the title ~~Set up type hinting: C extension stubs + typing for serializing_producer + deserializing_consumer~~ Set up type hinting Oct 16, 2025

address mypy complaints

f756116

This comment has been minimized.

Sign in to view

revert some accidental doc change

8be650a

This comment has been minimized.

Sign in to view

fangnx changed the title ~~Set up type hinting~~ Set up type hinting: add missing types to Python files and C extension stubs Oct 20, 2025

fangnx requested a review from Copilot October 20, 2025 22:15

Copilot AI reviewed Oct 20, 2025

View reviewed changes

fangnx added 2 commits October 21, 2025 11:32

Merge remote-tracking branch 'origin/master' into typehinting-kafka

5cb030e

fix some suggestions by copilot

bce471b

This comment has been minimized.

Sign in to view

linter

62f5c5b

fangnx requested a review from Copilot October 21, 2025 18:00

Copilot AI reviewed Oct 21, 2025

View reviewed changes

fangnx mentioned this pull request Oct 21, 2025

Set up type hinting: add correct types and apply type fixes in schema-registry module #2107

Merged

2 tasks

MSeal requested changes Oct 21, 2025

View reviewed changes

fangnx added 3 commits October 22, 2025 12:59

address comments

49e1348

bug fix

b65d202

merge

e500135

This comment has been minimized.

Sign in to view

more fix and add missing method

d26fc41

fangnx commented Oct 22, 2025

View reviewed changes

linter fix

4df06ca

This comment has been minimized.

Sign in to view

MSeal approved these changes Oct 22, 2025

View reviewed changes

MSeal merged commit 51652e3 into master Oct 22, 2025
3 checks passed

MSeal deleted the typehinting-kafka branch October 22, 2025 22:06

	def poll(self, timeout: float = -1) -> Optional['Message']:
	def poll(self, timeout: float = -1) -> Optional[Message]:

	self.loop.call_soon_threadsafe(lambda: self.logger.log(args, *kwargs))
	self.loop.call_soon_threadsafe(self.logger.log, args, *kwargs)



	# The stub file for Message.set_key/set_value expects bytes, but after deserialization, these can be any type.
	# The stub should be updated to accept Any. See CodeQL rule and issue for details.

	def __init__(self, topic_partitions: List[TopicPartition] = []) -> None:
	def __init__(self, topic_partitions: Optional[List[TopicPartition]] = None) -> None:

Set up type hinting: add missing types to Python files and C extension stubs #2041

Set up type hinting: add missing types to Python files and C extension stubs #2041

Uh oh!

Conversation

fangnx commented Sep 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

Checklist

References

Test & Review

Open questions / Follow-ups

Uh oh!

confluent-cla-assistant bot commented Sep 5, 2025

Uh oh!

This comment was marked as outdated.

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Sep 5, 2025

Choose a reason for hiding this comment

Uh oh!

This comment has been minimized.

MSeal left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This comment has been minimized.

This comment has been minimized.

This comment has been minimized.

This comment has been minimized.

This comment has been minimized.

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Copilot AI Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Copilot AI Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

This comment has been minimized.

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Uh oh!

Copilot AI Oct 21, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 21, 2025

Choose a reason for hiding this comment

Uh oh!

MSeal left a comment

fangnx commented Sep 5, 2025 •

edited

Loading