fix

Author: fangnx

src/confluent_kafka/admin/__init__.py

@@ -652,8 +652,7 @@ def list_topics(self, *args: Any, **kwargs: Any) -> ClusterMetadata:
 
         return super(AdminClient, self).list_topics(*args, **kwargs)
 
-    def list_groups(self, *args: Any, **kwargs: Any) -> GroupMetadata:
-
+    def list_groups(self, *args: Any, **kwargs: Any) -> List[GroupMetadata]:
         return super(AdminClient, self).list_groups(*args, **kwargs)
 
     def create_partitions(  # type: ignore[override]

src/confluent_kafka/admin/_metadata.py

@@ -12,7 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Dict, List
+from typing import Dict, List, Optional
+
+from confluent_kafka.cimpl import KafkaError
 
 
 class ClusterMetadata(object):

src/confluent_kafka/cimpl.pyi

@@ -38,6 +38,8 @@ from typing import Any, Optional, Callable, List, Tuple, Dict, Union, overload
 from typing_extensions import Self, Literal
 import builtins
 
+from confluent_kafka.admin._metadata import ClusterMetadata, GroupMetadata
+
 from ._types import HeadersType
 
 # Callback types with proper class references (defined locally to avoid circular imports)
@@ -199,6 +201,11 @@ class Consumer:
         message: Optional['Message'] = None,
         offsets: Optional[List[TopicPartition]] = None
     ) -> None: ...
+    def committed(
+        self,
+        partitions: List[TopicPartition],
+        timeout: float = -1
+    ) -> List[TopicPartition]: ...
     def close(self) -> None: ...
     def list_topics(self, topic: Optional[str] = None, timeout: float = -1) -> Any: ...
     def offsets_for_times(
@@ -255,12 +262,12 @@ class _AdminClientImpl:
         self,
         topic: Optional[str] = None,
         timeout: float = -1
-    ) -> Any: ...
+    ) -> ClusterMetadata: ...
     def list_groups(
         self,
         group: Optional[str] = None,
         timeout: float = -1
-    ) -> Any: ...
+    ) -> List[GroupMetadata]: ...
     def describe_consumer_groups(
         self,
         group_ids: List[str],

src/confluent_kafka/experimental/aio/_common.py

@@ -32,7 +32,7 @@ def __init__(
         self.logger = logger
 
     def log(self, *args: Any, **kwargs: Any) -> None:
-        self.loop.call_soon_threadsafe(self.logger.log, *args, **kwargs)
+        self.loop.call_soon_threadsafe(lambda: self.logger.log(*args, **kwargs))
 
 
 def wrap_callback(

src/confluent_kafka/serialization/__init__.py

@@ -17,7 +17,7 @@
 #
 import struct as _struct
 from enum import Enum
-from typing import Any, Optional
+from typing import Any, List, Optional
 
 from confluent_kafka.error import KafkaException
 from confluent_kafka._types import HeadersType
@@ -114,7 +114,7 @@ class Serializer(object):
           - unicode(encoding)
     """
 
-    __slots__ = []
+    __slots__: List[str] = []
 
     def __call__(self, obj: Any, ctx: Optional[SerializationContext] = None) -> Optional[bytes]:
         """