diff --git a/docs/index.rst b/docs/index.rst
index 8b59fdf99..459b0d814 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -1074,8 +1074,10 @@ addition to the properties dictated by the underlying librdkafka C library:
   commit error, or None on success. *list(TopicPartition)* is the list of partitions with their committed
   offsets or per-partition errors.
 
-* ``logger=logging.Handler`` kwarg: forward logs from the Kafka client to the
-  provided ``logging.Handler`` instance.
+* ``logger=logging.Logger`` kwarg: forward logs from the Kafka client to the
+  provided ``logging.Logger`` instance.
+  The client calls ``logger.log(level, msg, *args)`` internally, which matches
+  the ``logging.Logger`` interface (not ``logging.Handler``).
   To avoid spontaneous calls from non-Python threads the log messages
   will only be forwarded when ``client.poll()`` or ``producer.flush()`` are called.
   For example:
diff --git a/src/confluent_kafka/cimpl.pyi b/src/confluent_kafka/cimpl.pyi
index b8be621da..09cc8fa22 100644
--- a/src/confluent_kafka/cimpl.pyi
+++ b/src/confluent_kafka/cimpl.pyi
@@ -35,6 +35,7 @@ maintenance burden and get type hints directly from the implementation.
 """
 
 import builtins
+import logging
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union, overload
 
 from typing_extensions import Literal, Self
@@ -325,6 +326,8 @@ class Producer:
         self,
         config: Dict[str, Any],
         /,
+        *,
+        logger: Optional[logging.Logger] = None,
         **kwargs: Any
     ) -> None:
         """
@@ -333,6 +336,10 @@ class Producer:
 
         Args:
             config: Configuration dictionary.
+            logger: Optional ``logging.Logger`` instance to forward Kafka client
+                log messages to. The client calls ``logger.log(level, msg, *args)``
+                internally. Log messages are only forwarded when
+                ``producer.poll()`` or ``producer.flush()`` are called.
             **kwargs: Additional config as keyword args (overrides dict values).
 
         Example:
@@ -341,11 +348,15 @@ class Producer:
         """
         ...
     @overload
-    def __init__(self, **config: Any) -> None:
+    def __init__(self, *, logger: Optional[logging.Logger] = None, **config: Any) -> None:
         """
         Create Producer with keyword arguments only.
 
         Args:
+            logger: Optional ``logging.Logger`` instance to forward Kafka client
+                log messages to. The client calls ``logger.log(level, msg, *args)``
+                internally. Log messages are only forwarded when
+                ``producer.poll()`` or ``producer.flush()`` are called.
             **config: Configuration as keyword args.
                       Note: Use underscores (bootstrap_servers) not dots (bootstrap.servers) in kwargs.
 
@@ -408,6 +419,8 @@ class Consumer:
         self,
         config: dict[str, Any],
         /,
+        *,
+        logger: Optional[logging.Logger] = None,
         **kwargs: Any
     ) -> None:
         """
@@ -416,6 +429,10 @@ class Consumer:
 
         Args:
             config: Configuration dictionary. Must include 'group.id'.
+            logger: Optional ``logging.Logger`` instance to forward Kafka client
+                log messages to. The client calls ``logger.log(level, msg, *args)``
+                internally. Log messages are only forwarded when
+                ``consumer.poll()`` is called.
             **kwargs: Additional config as keyword args (overrides dict values).
 
         Example:
@@ -424,11 +441,15 @@ class Consumer:
         """
         ...
     @overload
-    def __init__(self, **config: Any) -> None:
+    def __init__(self, *, logger: Optional[logging.Logger] = None, **config: Any) -> None:
         """
         Create Consumer with keyword arguments only.
 
         Args:
+            logger: Optional ``logging.Logger`` instance to forward Kafka client
+                log messages to. The client calls ``logger.log(level, msg, *args)``
+                internally. Log messages are only forwarded when
+                ``consumer.poll()`` is called.
             **config: Configuration as keyword args. Must include group_id.
                       Note: Use underscores (group_id) not dots (group.id) in kwargs.