Repository: airtai/faststream
Pull request #2090 (2025-03-04 18:26:13)

Add delay to unassigned consumer warning
# Description

I've noticed that a rolling update that temporariry creates an excess of application instances (e.g. if surge > unavailable) will trigger the unassigned consumer warnings unneccessarily.

For example, if normally there are 2 instances with 5 consumers each reading from 10 partitions, an update might create a 3rd instance with an additional 5 consumers before shuttinng down on of the older instances. That would make 5 consumers spread across the 3 instances temporarily unassigned.

I've added a 5 minute delay before a warning. If the consumer becomes assigned during that delay the warning is not triggered. That would allow an update to finish without triggering the warning.

## Type of change

Please delete options that are not relevant.

- [x] Documentation (typos, code examples, or any documentation updates)
- [ ] Bug fix (a non-breaking change that resolves an issue)
- [x] New feature (a non-breaking change that adds functionality)
- [ ] Breaking change (a fix or feature that would disrupt existing functionality)
- [ ] This change requires a documentation update

## Checklist

- [x] My code adheres to the style guidelines of this project (`scripts/lint.sh` shows no errors)
- [x] I have conducted a self-review of my own code
- [x] I have made the necessary changes to the documentation
- [x] My changes do not generate any new warnings
- [x] I have added tests to validate the effectiveness of my fix or the functionality of my new feature
- [x] Both new and existing unit tests pass successfully on my local environment by running `scripts/test-cov.sh`
- [x] I have ensured that static analysis tests are passing by running `scripts/static-analysis.sh`
- [ ] I have included code examples to illustrate the modifications



Patch:
diff --git a/docs/docs/SUMMARY.md b/docs/docs/SUMMARY.md
index ebb3c44a..ce6ca266 100644
--- a/docs/docs/SUMMARY.md
+++ b/docs/docs/SUMMARY.md
@@ -564,6 +564,7 @@ search:
                     - [TopicPartition](api/faststream/confluent/schemas/TopicPartition.md)
                     - params
                         - [ConsumerConnectionParams](api/faststream/confluent/schemas/params/ConsumerConnectionParams.md)
+                        - [SecurityOptions](api/faststream/confluent/schemas/params/SecurityOptions.md)
                     - partition
                         - [TopicPartition](api/faststream/confluent/schemas/partition/TopicPartition.md)
                 - security
diff --git a/docs/docs/en/api/faststream/confluent/schemas/params/SecurityOptions.md b/docs/docs/en/api/faststream/confluent/schemas/params/SecurityOptions.md
new file mode 100644
index 00000000..d8e40ac4
--- /dev/null
+++ b/docs/docs/en/api/faststream/confluent/schemas/params/SecurityOptions.md
@@ -0,0 +1,11 @@
+---
+# 0.5 - API
+# 2 - Release
+# 3 - Contributing
+# 5 - Template Page
+# 10 - Default
+search:
+  boost: 0.5
+---
+
+::: faststream.confluent.schemas.params.SecurityOptions
diff --git a/docs/docs/en/kafka/Subscriber/index.md b/docs/docs/en/kafka/Subscriber/index.md
index 8e0f4c51..1757c363 100644
--- a/docs/docs/en/kafka/Subscriber/index.md
+++ b/docs/docs/en/kafka/Subscriber/index.md
@@ -77,5 +77,5 @@ async def base_handler(
 ## Concurrent processing
 
 There are two possible modes of concurrent message processing:
-- With `auto_commit=False` and `max_workers` > 1, a handler processes all messages concurrently in a at-most-once semantic.
-- With `auto_commit=True` and `max_workers` > 1, processing is concurrent between topic partitions and sequential within a partition to ensure reliable at-least-once processing. Maximum concurrency is achieved when total number of workers across all application instances running workers in the same consumer group is equal to the number of partitions in the topic. Increasing worker count beyond that will result in idle workers as not more than one consumer from a consumer group can be consuming from the same partition.
+* With `auto_commit=False` and `max_workers` > 1, a handler processes all messages concurrently in a at-most-once semantic.
+* With `auto_commit=True` and `max_workers` > 1, processing is concurrent between topic partitions and sequential within a partition to ensure reliable at-least-once processing. Maximum concurrency is achieved when total number of workers across all application instances running workers in the same consumer group is equal to the number of partitions in the topic. Increasing worker count beyond that will result in idle workers as not more than one consumer from a consumer group can be consuming from the same partition.
diff --git a/faststream/__about__.py b/faststream/__about__.py
index 9f10e18e..a0fb0711 100644
--- a/faststream/__about__.py
+++ b/faststream/__about__.py
@@ -1,5 +1,5 @@
 """Simple and fast framework to create message brokers based microservices."""
 
-__version__ = "0.5.35"
+__version__ = "0.5.36"
 
 SERVICE_NAME = f"faststream-{__version__}"
diff --git a/faststream/confluent/client.py b/faststream/confluent/client.py
index eb8ce40a..666f5670 100644
--- a/faststream/confluent/client.py
+++ b/faststream/confluent/client.py
@@ -29,6 +29,7 @@ from faststream.utils.functions import call_or_await
 if TYPE_CHECKING:
     from typing_extensions import NotRequired, TypedDict
 
+    from faststream.confluent.schemas.params import SecurityOptions
     from faststream.types import AnyDict, LoggerProto
 
     class _SendKwargs(TypedDict):
@@ -64,9 +65,7 @@ class AsyncConfluentProducer:
         transactional_id: Optional[Union[str, int]] = None,
         transaction_timeout_ms: int = 60000,
         allow_auto_create_topics: bool = True,
-        sasl_mechanism: Optional[str] = None,
-        sasl_plain_password: Optional[str] = None,
-        sasl_plain_username: Optional[str] = None,
+        security_config: Optional["SecurityOptions"] = None,
     ) -> None:
         self.logger = logger
 
@@ -101,16 +100,11 @@ class AsyncConfluentProducer:
             "allow.auto.create.topics": allow_auto_create_topics,
         }
 
-        final_config = {**config.as_config_dict(), **config_from_params}
-
-        if sasl_mechanism in ["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"]:
-            final_config.update(
-                {
-                    "sasl.mechanism": sasl_mechanism,
-                    "sasl.username": sasl_plain_username,
-                    "sasl.password": sasl_plain_password,
-                }
-            )
+        final_config = {
+            **config.as_config_dict(),
+            **config_from_params,
+            **dict(security_config or {}),
+        }
 
         self.producer = Producer(final_config, logger=self.logger)  # type: ignore[call-arg]
 
@@ -247,9 +241,7 @@ class AsyncConfluentConsumer:
         connections_max_idle_ms: int = 540000,
         isolation_level: str = "read_uncommitted",
         allow_auto_create_topics: bool = True,
-        sasl_mechanism: Optional[str] = None,
-        sasl_plain_password: Optional[str] = None,
-        sasl_plain_username: Optional[str] = None,
+        security_config: Optional["SecurityOptions"] = None,
     ) -> None:
         self.logger = logger
 
@@ -284,7 +276,6 @@ class AsyncConfluentConsumer:
             "fetch.max.bytes": fetch_max_bytes,
             "fetch.min.bytes": fetch_min_bytes,
             "max.partition.fetch.bytes": max_partition_fetch_bytes,
-            # "request.timeout.ms": 1000,  # producer only
             "fetch.error.backoff.ms": retry_backoff_ms,
             "auto.offset.reset": auto_offset_reset,
             "enable.auto.commit": enable_auto_commit,
@@ -301,15 +292,7 @@ class AsyncConfluentConsumer:
         }
         self.allow_auto_create_topics = allow_auto_create_topics
         final_config.update(config_from_params)
-
-        if sasl_mechanism in ["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"]:
-            final_config.update(
-                {
-                    "sasl.mechanism": sasl_mechanism,
-                    "sasl.username": sasl_plain_username,
-                    "sasl.password": sasl_plain_password,
-                }
-            )
+        final_config.update(**dict(security_config or {}))
 
         self.config = final_config
         self.consumer = Consumer(final_config, logger=self.logger)  # type: ignore[call-arg]
diff --git a/faststream/confluent/schemas/params.py b/faststream/confluent/schemas/params.py
index 91cb8359..e24a6ad4 100644
--- a/faststream/confluent/schemas/params.py
+++ b/faststream/confluent/schemas/params.py
@@ -1,6 +1,20 @@
 from typing import List, Literal, Union
 
-from typing_extensions import TypedDict
+from typing_extensions import Required, TypedDict
+
+
+class SecurityOptions(TypedDict, total=False):
+    sasl_mechanism: Required[
+        Literal[
+            "PLAIN",
+            "GSSAPI",
+            "SCRAM-SHA-256",
+            "SCRAM-SHA-512",
+            "OAUTHBEARER",
+        ]
+    ]
+    sasl_plain_password: str
+    sasl_plain_username: str
 
 
 class ConsumerConnectionParams(TypedDict, total=False):
@@ -16,12 +30,4 @@ class ConsumerConnectionParams(TypedDict, total=False):
     ]
     connections_max_idle_ms: int
     allow_auto_create_topics: bool
-    sasl_mechanism: Literal[
-        "PLAIN",
-        "GSSAPI",
-        "SCRAM-SHA-256",
-        "SCRAM-SHA-512",
-        "OAUTHBEARER",
-    ]
-    sasl_plain_password: str
-    sasl_plain_username: str
+    security_config: SecurityOptions
diff --git a/faststream/confluent/security.py b/faststream/confluent/security.py
index 4e3e0c0e..6324b4c4 100644
--- a/faststream/confluent/security.py
+++ b/faststream/confluent/security.py
@@ -48,39 +48,49 @@ def _parse_base_security(security: BaseSecurity) -> "AnyDict":
 def _parse_sasl_plaintext(security: SASLPlaintext) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "PLAIN",
-        "sasl_plain_username": security.username,
-        "sasl_plain_password": security.password,
+        "security_config": {
+            "sasl_mechanism": "PLAIN",
+            "sasl_plain_username": security.username,
+            "sasl_plain_password": security.password,
+        },
     }
 
 
 def _parse_sasl_scram256(security: SASLScram256) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "SCRAM-SHA-256",
-        "sasl_plain_username": security.username,
-        "sasl_plain_password": security.password,
+        "security_config": {
+            "sasl_mechanism": "SCRAM-SHA-256",
+            "sasl_plain_username": security.username,
+            "sasl_plain_password": security.password,
+        },
     }
 
 
 def _parse_sasl_scram512(security: SASLScram512) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "SCRAM-SHA-512",
-        "sasl_plain_username": security.username,
-        "sasl_plain_password": security.password,
+        "security_config": {
+            "sasl_mechanism": "SCRAM-SHA-512",
+            "sasl_plain_username": security.username,
+            "sasl_plain_password": security.password,
+        },
     }
 
 
 def _parse_sasl_oauthbearer(security: SASLOAuthBearer) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "OAUTHBEARER",
+        "security_config": {
+            "sasl_mechanism": "OAUTHBEARER",
+        },
     }
 
 
 def _parse_sasl_gssapi(security: SASLGSSAPI) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
-        "sasl_mechanism": "GSSAPI",
+        "security_config": {
+            "sasl_mechanism": "GSSAPI",
+        },
     }
diff --git a/faststream/kafka/listener.py b/faststream/kafka/listener.py
index b6c8c8d7..d70e6f19 100644
--- a/faststream/kafka/listener.py
+++ b/faststream/kafka/listener.py
@@ -13,6 +13,8 @@ if TYPE_CHECKING:
 class LoggingListenerProxy(ConsumerRebalanceListener):  # type: ignore[misc]
     """Logs partition assignments and passes calls to user-supplied listener."""
 
+    _log_unassigned_consumer_delay_seconds = 60 * 2
+
     def __init__(
         self,
         consumer: "AIOKafkaConsumer",
@@ -22,6 +24,17 @@ class LoggingListenerProxy(ConsumerRebalanceListener):  # type: ignore[misc]
         self.consumer = consumer
         self.logger = logger
         self.listener = listener
+        self._log_unassigned_consumer_task: Optional[asyncio.Task[None]] = None
+
+    async def log_unassigned_consumer(self) -> None:
+        await asyncio.sleep(self._log_unassigned_consumer_delay_seconds)
+        self._log(
+            logging.WARNING,
+            f"Consumer in group {self.consumer._group_id} has had no partition "
+            f"assignments for {self._log_unassigned_consumer_delay_seconds} seconds: "
+            f"topics {self.consumer._subscription.topics} may have fewer partitions "
+            f"than consumers.",
+        )
 
     async def on_partitions_revoked(self, revoked: Set["TopicPartition"]) -> None:
         if self.listener:
@@ -38,9 +51,16 @@ class LoggingListenerProxy(ConsumerRebalanceListener):  # type: ignore[misc]
         if not assigned:
             self._log(
                 logging.WARNING,
-                f"Consumer in group {self.consumer._group_id} has no partition assignments - topics "
-                f"{self.consumer._subscription.topics} may have fewer partitions than consumers",
+                f"Consumer in group {self.consumer._group_id} has no partition assignments - this "
+                f"could be temporary, e.g. during a rolling update. A separate warning will be logged if "
+                f"this condition persists for {self._log_unassigned_consumer_delay_seconds} seconds.",
+            )
+            self._log_unassigned_consumer_task = asyncio.create_task(
+                self.log_unassigned_consumer()
             )
+        elif self._log_unassigned_consumer_task:
+            self._log_unassigned_consumer_task.cancel()
+            self._log_unassigned_consumer_task = None
 
         if self.listener:
             call_result = self.listener.on_partitions_assigned(assigned)


Test patch:
diff --git a/tests/a_docs/confluent/test_security.py b/tests/a_docs/confluent/test_security.py
index b77c91eb..b0ba4c86 100644
--- a/tests/a_docs/confluent/test_security.py
+++ b/tests/a_docs/confluent/test_security.py
@@ -28,11 +28,14 @@ async def test_scram256():
         async with scram256_broker:
             producer_call_kwargs = producer.call_args.kwargs
 
-            call_kwargs = {}
-            call_kwargs["sasl_mechanism"] = "SCRAM-SHA-256"
-            call_kwargs["sasl_plain_username"] = "admin"
-            call_kwargs["sasl_plain_password"] = "password"  # pragma: allowlist secret
-            call_kwargs["security_protocol"] = "SASL_SSL"
+            call_kwargs = {
+                "security_config": {
+                    "sasl_mechanism": "SCRAM-SHA-256",
+                    "sasl_plain_username": "admin",
+                    "sasl_plain_password": "password",  # pragma: allowlist secret
+                },
+                "security_protocol": "SASL_SSL",
+            }
 
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
@@ -53,11 +56,14 @@ async def test_scram512():
         async with scram512_broker:
             producer_call_kwargs = producer.call_args.kwargs
 
-            call_kwargs = {}
-            call_kwargs["sasl_mechanism"] = "SCRAM-SHA-512"
-            call_kwargs["sasl_plain_username"] = "admin"
-            call_kwargs["sasl_plain_password"] = "password"  # pragma: allowlist secret
-            call_kwargs["security_protocol"] = "SASL_SSL"
+            call_kwargs = {
+                "security_config": {
+                    "sasl_mechanism": "SCRAM-SHA-512",
+                    "sasl_plain_username": "admin",
+                    "sasl_plain_password": "password",  # pragma: allowlist secret
+                },
+                "security_protocol": "SASL_SSL",
+            }
 
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
@@ -78,11 +84,14 @@ async def test_plaintext():
         async with plaintext_broker:
             producer_call_kwargs = producer.call_args.kwargs
 
-            call_kwargs = {}
-            call_kwargs["sasl_mechanism"] = "PLAIN"
-            call_kwargs["sasl_plain_username"] = "admin"
-            call_kwargs["sasl_plain_password"] = "password"  # pragma: allowlist secret
-            call_kwargs["security_protocol"] = "SASL_SSL"
+            call_kwargs = {
+                "security_config": {
+                    "sasl_mechanism": "PLAIN",
+                    "sasl_plain_username": "admin",
+                    "sasl_plain_password": "password",  # pragma: allowlist secret
+                },
+                "security_protocol": "SASL_SSL",
+            }
 
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
@@ -103,9 +112,12 @@ async def test_oathbearer():
         async with oauthbearer_broker:
             producer_call_kwargs = producer.call_args.kwargs
 
-            call_kwargs = {}
-            call_kwargs["sasl_mechanism"] = "OAUTHBEARER"
-            call_kwargs["security_protocol"] = "SASL_SSL"
+            call_kwargs = {
+                "security_config": {
+                    "sasl_mechanism": "OAUTHBEARER",
+                },
+                "security_protocol": "SASL_SSL",
+            }
 
             assert call_kwargs.items() <= producer_call_kwargs.items()
 
@@ -127,7 +139,9 @@ async def test_gssapi():
             producer_call_kwargs = producer.call_args.kwargs
 
             call_kwargs = {
-                "sasl_mechanism": "GSSAPI",
+                "security_config": {
+                    "sasl_mechanism": "GSSAPI",
+                },
                 "security_protocol": "SASL_SSL",
             }
 
diff --git a/tests/brokers/kafka/test_consume.py b/tests/brokers/kafka/test_consume.py
index 3b034b75..c30a989d 100644
--- a/tests/brokers/kafka/test_consume.py
+++ b/tests/brokers/kafka/test_consume.py
@@ -9,6 +9,7 @@ from aiokafka.admin import AIOKafkaAdminClient, NewTopic
 from faststream.exceptions import AckMessage
 from faststream.kafka import KafkaBroker, TopicPartition
 from faststream.kafka.annotations import KafkaMessage
+from faststream.kafka.listener import LoggingListenerProxy
 from tests.brokers.base.consume import BrokerRealConsumeTestcase
 from tests.tools import spy_decorator
 
@@ -472,7 +473,12 @@ class TestConsume(BrokerRealConsumeTestcase):
         queue: str,
         partitions: int,
         warning: bool,
+        monkeypatch: pytest.MonkeyPatch,
     ):
+        monkeypatch.setattr(
+            LoggingListenerProxy, "_log_unassigned_consumer_delay_seconds", 0
+        )
+
         admin_client = AIOKafkaAdminClient()
         try:
             await admin_client.start()
@@ -504,7 +510,7 @@ class TestConsume(BrokerRealConsumeTestcase):
                             if x[0][0] == logging.WARNING
                         ]
                     )
-                    == 1
+                    == 2
                 )
             else:
                 assert (