Repository: airtai/faststream (python)

Issue:
Bug: confluent.KafkaRouter with SASLScram512 fails in version 0.5.36
**Describe the bug**
From https://discord.com/channels/1085457301214855171/1085457302280228898/1352183480049729566:

After updating to version faststream==0.5.36, we’re experiencing an issue with confluent.fastapi.KafkaRouter on startup. Our dependencies: faststream==0.5.36, confluent-kafka==2.8.0.

We rolled back to version 0.5.35, and the issue disappeared.

**How to reproduce**
Include source code:

```python
security = SASLScram512(
    username=KAFKA_SETTINGS.kafka_auth_user,
    password=KAFKA_SETTINGS.kafka_auth_password,
    use_ssl=False,
)
router = KafkaRouter(
    KAFKA_SETTINGS.kafka_instance,
    security=security,
)
```

**Observed behavior**
```txt
File "/usr/local/lib/python3.8/site-packages/faststream/broker/fastapi/router.py", line 324, in start_broker_lifespan
    await self.broker.start()
File "/usr/local/lib/python3.8/site-packages/faststream/confluent/broker/broker.py", line 456, in start
    await super().start()
File "/usr/local/lib/python3.8/site-packages/faststream/broker/core/usecase.py", line 219, in start
    await self.connect()
File "/usr/local/lib/python3.8/site-packages/faststream/confluent/broker/broker.py", line 423, in connect
    return await super().connect(**kwargs)
File "/usr/local/lib/python3.8/site-packages/faststream/broker/core/usecase.py", line 226, in connect
    self._connection = await self._connect(**connection_kwargs)
File "/usr/local/lib/python3.8/site-packages/faststream/confluent/broker/broker.py", line 435, in _connect
    native_producer = AsyncConfluentProducer(
File "/usr/local/lib/python3.8/site-packages/faststream/confluent/client.py", line 109, in __init__
    self.producer = Producer(final_config, logger=self.logger) # type: ignore[call-arg]
cimpl.KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="No such configuration property: "sasl_mechanism""}
```

**Environment**
Our dependencies: faststream==0.5.36, confluent-kafka==2.8.0.



Pull request:
fix: correct Confluent options name mapping
# Description

Please include a summary of the change and specify which issue is being addressed. Additionally, provide relevant motivation and context.

Fixes #2136 

## Type of change

Please delete options that are not relevant.

- [ ] Documentation (typos, code examples, or any documentation updates)
- [x] Bug fix (a non-breaking change that resolves an issue)
- [ ] 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

- [ ] My code adheres to the style guidelines of this project (`scripts/lint.sh` shows no errors)
- [ ] I have conducted a self-review of my own code
- [ ] I have made the necessary changes to the documentation
- [ ] My changes do not generate any new warnings
- [ ] I have added tests to validate the effectiveness of my fix or the functionality of my new feature
- [ ] Both new and existing unit tests pass successfully on my local environment by running `scripts/test-cov.sh`
- [ ] 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/faststream/__about__.py b/faststream/__about__.py
index a0fb0711..4e07cf76 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.36"
+__version__ = "0.5.37"
 
 SERVICE_NAME = f"faststream-{__version__}"
diff --git a/faststream/broker/subscriber/call_item.py b/faststream/broker/subscriber/call_item.py
index 5f4f56ea..dfd68815 100644
--- a/faststream/broker/subscriber/call_item.py
+++ b/faststream/broker/subscriber/call_item.py
@@ -9,6 +9,7 @@ from typing import (
     Generic,
     Iterable,
     Optional,
+    Reversible,
     Sequence,
     cast,
 )
@@ -81,7 +82,7 @@ class HandlerItem(SetupAble, Generic[MsgType]):
         apply_types: bool,
         is_validate: bool,
         _get_dependant: Optional[Callable[..., Any]],
-        _call_decorators: Iterable["Decorator"],
+        _call_decorators: Reversible["Decorator"],
     ) -> None:
         if self.dependant is None:
             self.item_parser = parser
diff --git a/faststream/broker/wrapper/call.py b/faststream/broker/wrapper/call.py
index 30725476..c4c1cb78 100644
--- a/faststream/broker/wrapper/call.py
+++ b/faststream/broker/wrapper/call.py
@@ -9,6 +9,7 @@ from typing import (
     List,
     Mapping,
     Optional,
+    Reversible,
     Sequence,
     Union,
 )
@@ -152,10 +153,10 @@ class HandlerCallWrapper(Generic[MsgType, P_HandlerParams, T_HandlerReturn]):
         is_validate: bool,
         dependencies: Iterable["Depends"],
         _get_dependant: Optional[Callable[..., Any]],
-        _call_decorators: Iterable["Decorator"],
+        _call_decorators: Reversible["Decorator"],
     ) -> Optional["CallModel[..., Any]"]:
         call = self._original_call
-        for decor in _call_decorators:
+        for decor in reversed(_call_decorators):
             call = decor(call)
         self._original_call = call
 
diff --git a/faststream/confluent/schemas/params.py b/faststream/confluent/schemas/params.py
index e24a6ad4..cdde520f 100644
--- a/faststream/confluent/schemas/params.py
+++ b/faststream/confluent/schemas/params.py
@@ -2,19 +2,23 @@ from typing import List, Literal, Union
 
 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
+SecurityOptions = TypedDict(
+    "SecurityOptions",
+    {
+        "sasl.mechanism": Required[
+            Literal[
+                "PLAIN",
+                "GSSAPI",
+                "SCRAM-SHA-256",
+                "SCRAM-SHA-512",
+                "OAUTHBEARER",
+            ]
+        ],
+        "sasl.password": str,
+        "sasl.username": str,
+    },
+    total=False,
+)
 
 
 class ConsumerConnectionParams(TypedDict, total=False):
diff --git a/faststream/confluent/security.py b/faststream/confluent/security.py
index 6324b4c4..3eef59cb 100644
--- a/faststream/confluent/security.py
+++ b/faststream/confluent/security.py
@@ -49,9 +49,9 @@ def _parse_sasl_plaintext(security: SASLPlaintext) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
         "security_config": {
-            "sasl_mechanism": "PLAIN",
-            "sasl_plain_username": security.username,
-            "sasl_plain_password": security.password,
+            "sasl.mechanism": "PLAIN",
+            "sasl.username": security.username,
+            "sasl.password": security.password,
         },
     }
 
@@ -60,9 +60,9 @@ def _parse_sasl_scram256(security: SASLScram256) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
         "security_config": {
-            "sasl_mechanism": "SCRAM-SHA-256",
-            "sasl_plain_username": security.username,
-            "sasl_plain_password": security.password,
+            "sasl.mechanism": "SCRAM-SHA-256",
+            "sasl.username": security.username,
+            "sasl.password": security.password,
         },
     }
 
@@ -71,9 +71,9 @@ def _parse_sasl_scram512(security: SASLScram512) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
         "security_config": {
-            "sasl_mechanism": "SCRAM-SHA-512",
-            "sasl_plain_username": security.username,
-            "sasl_plain_password": security.password,
+            "sasl.mechanism": "SCRAM-SHA-512",
+            "sasl.username": security.username,
+            "sasl.password": security.password,
         },
     }
 
@@ -82,7 +82,7 @@ def _parse_sasl_oauthbearer(security: SASLOAuthBearer) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
         "security_config": {
-            "sasl_mechanism": "OAUTHBEARER",
+            "sasl.mechanism": "OAUTHBEARER",
         },
     }
 
@@ -91,6 +91,6 @@ def _parse_sasl_gssapi(security: SASLGSSAPI) -> "AnyDict":
     return {
         "security_protocol": "SASL_SSL" if security.use_ssl else "SASL_PLAINTEXT",
         "security_config": {
-            "sasl_mechanism": "GSSAPI",
+            "sasl.mechanism": "GSSAPI",
         },
     }


Test patch:
diff --git a/tests/a_docs/confluent/test_security.py b/tests/a_docs/confluent/test_security.py
index b0ba4c86..bba3b12d 100644
--- a/tests/a_docs/confluent/test_security.py
+++ b/tests/a_docs/confluent/test_security.py
@@ -30,9 +30,9 @@ async def test_scram256():
 
             call_kwargs = {
                 "security_config": {
-                    "sasl_mechanism": "SCRAM-SHA-256",
-                    "sasl_plain_username": "admin",
-                    "sasl_plain_password": "password",  # pragma: allowlist secret
+                    "sasl.mechanism": "SCRAM-SHA-256",
+                    "sasl.username": "admin",
+                    "sasl.password": "password",  # pragma: allowlist secret
                 },
                 "security_protocol": "SASL_SSL",
             }
@@ -58,9 +58,9 @@ async def test_scram512():
 
             call_kwargs = {
                 "security_config": {
-                    "sasl_mechanism": "SCRAM-SHA-512",
-                    "sasl_plain_username": "admin",
-                    "sasl_plain_password": "password",  # pragma: allowlist secret
+                    "sasl.mechanism": "SCRAM-SHA-512",
+                    "sasl.username": "admin",
+                    "sasl.password": "password",  # pragma: allowlist secret
                 },
                 "security_protocol": "SASL_SSL",
             }
@@ -86,9 +86,9 @@ async def test_plaintext():
 
             call_kwargs = {
                 "security_config": {
-                    "sasl_mechanism": "PLAIN",
-                    "sasl_plain_username": "admin",
-                    "sasl_plain_password": "password",  # pragma: allowlist secret
+                    "sasl.mechanism": "PLAIN",
+                    "sasl.username": "admin",
+                    "sasl.password": "password",  # pragma: allowlist secret
                 },
                 "security_protocol": "SASL_SSL",
             }
@@ -114,7 +114,7 @@ async def test_oathbearer():
 
             call_kwargs = {
                 "security_config": {
-                    "sasl_mechanism": "OAUTHBEARER",
+                    "sasl.mechanism": "OAUTHBEARER",
                 },
                 "security_protocol": "SASL_SSL",
             }
@@ -140,7 +140,7 @@ async def test_gssapi():
 
             call_kwargs = {
                 "security_config": {
-                    "sasl_mechanism": "GSSAPI",
+                    "sasl.mechanism": "GSSAPI",
                 },
                 "security_protocol": "SASL_SSL",
             }