Repository: airtai/faststream (python)

Issue:
Bug: Duplicate log messages when starting multiple broker instances due to repeated stdout handler registration
**Describe the bug**
While developing an application and writing integration tests using this library, I observed the following logs. Although each test case creates a new broker instance and calls `start()` and `close()` to release resources, the log output shows that one `consumer handler` was registered multiple times.

![Image](https://github.com/user-attachments/assets/5e0fed19-22c3-429f-a635-3bf2c445395e)


As I traced through the code, I noticed that the `consumer handler` was only called once per message. After analyzing the library's code, I found that this behavior was caused by **log messages being duplicated**, rather than consumers being duplicated.

This duplication occurs because the `sys.stdout` handler is added to the logger multiple times each time a broker is started, resulting in repeated log outputs. 
- Even if I create **a new broker instance**, the logs are duplicated.


**How to reproduce**

```python
import asyncio  
from faststream import Logger  
from faststream.rabbit import RabbitBroker, RabbitExchange, RabbitQueue  
  
  
def create_new_broker(event: asyncio.Event):  
    broker = RabbitBroker( max_consumers=100, reconnect_interval=5)  
    exch = RabbitExchange("exchange", auto_delete=True)  
    queue_1 = RabbitQueue("test-q-1", auto_delete=True)  
  
    @broker.subscriber(queue_1, exch)  
    async def base_handler1(logger: Logger):  
        logger.info("base_handler1")  
        event.set()  
    return broker  
  
  
async def test_case():  
    print("TEST START")  
    consumed_event = asyncio.Event()  
    broker = create_new_broker(consumed_event)  
    await broker.start()  
    await broker.publish(queue="test-q-1", exchange="exchange")  
    await consumed_event.wait()  
    await broker.close()  
    print("TEST END")  
  
  
async def main():  
    for i in range(3):  
        await test_case()  
  
  
if __name__ == "__main__":  
    loop = asyncio.get_event_loop()  
    loop.run_until_complete(main())
```


**Expected behavior**

Log messages should not be duplicated.

```
TEST START
2025-03-27 20:51:38,510 INFO     -          |          |            - Set max consumers to 100
2025-03-27 20:51:38,512 INFO     - exchange | test-q-1 |            - `BaseHandler1` waiting for messages
2025-03-27 20:51:38,522 INFO     - exchange | test-q-1 | 0ab47995aa - Received
2025-03-27 20:51:38,522 INFO     - exchange | test-q-1 | 0ab47995aa - base_handler1
2025-03-27 20:51:38,523 INFO     - exchange | test-q-1 | 0ab47995aa - Processed
TEST END
TEST START
2025-03-27 20:51:38,637 INFO     -          |          |            - Set max consumers to 100
2025-03-27 20:51:38,640 INFO     - exchange | test-q-1 |            - `BaseHandler1` waiting for messages
2025-03-27 20:51:38,650 INFO     - exchange | test-q-1 | 68e11b352f - Received
2025-03-27 20:51:38,650 INFO     - exchange | test-q-1 | 68e11b352f - base_handler1
2025-03-27 20:51:38,650 INFO     - exchange | test-q-1 | 68e11b352f - Processed
TEST END
TEST START
2025-03-27 20:51:38,757 INFO     -          |          |            - Set max consumers to 100
2025-03-27 20:51:38,758 INFO     - exchange | test-q-1 |            - `BaseHandler1` waiting for messages
2025-03-27 20:51:38,765 INFO     - exchange | test-q-1 | 887b8c61a7 - Received
2025-03-27 20:51:38,766 INFO     - exchange | test-q-1 | 887b8c61a7 - base_handler1
2025-03-27 20:51:38,766 INFO     - exchange | test-q-1 | 887b8c61a7 - Processed
TEST END
```

**Observed behavior**
As the test runs multiple times, log messages are duplicated, which could be mistakenly interpreted as multiple consumers being registered.

```
TEST START
2025-03-27 20:51:38,510 INFO     -          |          |            - Set max consumers to 100
2025-03-27 20:51:38,512 INFO     - exchange | test-q-1 |            - `BaseHandler1` waiting for messages
2025-03-27 20:51:38,522 INFO     - exchange | test-q-1 | 0ab47995aa - Received
2025-03-27 20:51:38,522 INFO     - exchange | test-q-1 | 0ab47995aa - base_handler1
2025-03-27 20:51:38,523 INFO     - exchange | test-q-1 | 0ab47995aa - Processed
TEST END
TEST START
2025-03-27 20:51:38,637 INFO     -          |          |            - Set max consumers to 100
2025-03-27 20:51:38,637 INFO     -          |          |            - Set max consumers to 100
2025-03-27 20:51:38,640 INFO     - exchange | test-q-1 |            - `BaseHandler1` waiting for messages
2025-03-27 20:51:38,640 INFO     - exchange | test-q-1 |            - `BaseHandler1` waiting for messages
2025-03-27 20:51:38,650 INFO     - exchange | test-q-1 | 68e11b352f - Received
2025-03-27 20:51:38,650 INFO     - exchange | test-q-1 | 68e11b352f - Received
2025-03-27 20:51:38,650 INFO     - exchange | test-q-1 | 68e11b352f - base_handler1
2025-03-27 20:51:38,650 INFO     - exchange | test-q-1 | 68e11b352f - base_handler1
2025-03-27 20:51:38,650 INFO     - exchange | test-q-1 | 68e11b352f - Processed
2025-03-27 20:51:38,650 INFO     - exchange | test-q-1 | 68e11b352f - Processed
TEST END
TEST START
2025-03-27 20:51:38,757 INFO     -          |          |            - Set max consumers to 100
2025-03-27 20:51:38,757 INFO     -          |          |            - Set max consumers to 100
2025-03-27 20:51:38,757 INFO     -          |          |            - Set max consumers to 100
2025-03-27 20:51:38,758 INFO     - exchange | test-q-1 |            - `BaseHandler1` waiting for messages
2025-03-27 20:51:38,758 INFO     - exchange | test-q-1 |            - `BaseHandler1` waiting for messages
2025-03-27 20:51:38,758 INFO     - exchange | test-q-1 |            - `BaseHandler1` waiting for messages
2025-03-27 20:51:38,765 INFO     - exchange | test-q-1 | 887b8c61a7 - Received
2025-03-27 20:51:38,765 INFO     - exchange | test-q-1 | 887b8c61a7 - Received
2025-03-27 20:51:38,765 INFO     - exchange | test-q-1 | 887b8c61a7 - Received
2025-03-27 20:51:38,766 INFO     - exchange | test-q-1 | 887b8c61a7 - base_handler1
2025-03-27 20:51:38,766 INFO     - exchange | test-q-1 | 887b8c61a7 - base_handler1
2025-03-27 20:51:38,766 INFO     - exchange | test-q-1 | 887b8c61a7 - base_handler1
2025-03-27 20:51:38,766 INFO     - exchange | test-q-1 | 887b8c61a7 - Processed
2025-03-27 20:51:38,766 INFO     - exchange | test-q-1 | 887b8c61a7 - Processed
2025-03-27 20:51:38,766 INFO     - exchange | test-q-1 | 887b8c61a7 - Processed
TEST END
```

**Screenshots**

![Image](https://github.com/user-attachments/assets/285301ca-4b74-4e58-b566-c9081d04e060)

**Environment**
Running FastStream 0.5.37 with CPython 3.12.9 on Windows 
Running FastStream 0.5.37 with CPython 3.11.7 on Darwin

---

Would it be alright if I submitted a pull request to address this issue?
Thanks for maintaining such a great library!


Pull request:
fix: prevent duplicate logging handler registration when calling broker.start()
# Description
Fixes #2149

This PR resolves #2149, which describes an issue where StreamHandler(sys.stdout) instances are redundantly added to the logger when the broker.start() function is called multiple times.

This change ensures that a StreamHandler is added only if an identical handler is not already present, preventing duplicate log messages.

The following example demonstrates the issue by running broker.start() in a loop:
```
import asyncio
from faststream import Logger
from faststream.rabbit import RabbitBroker, RabbitExchange, RabbitQueue


def create_new_broker(event: asyncio.Event):
    broker = RabbitBroker(max_consumers=100, reconnect_interval=5)
    exch = RabbitExchange("exchange", auto_delete=True)
    queue_1 = RabbitQueue("test-q-1", auto_delete=True)

    @broker.subscriber(queue_1, exch)
    async def base_handler1(logger: Logger):
        logger.info("base_handler1")
        event.set()

    return broker


async def test_case():
    print("TEST START")
    consumed_event = asyncio.Event()
    broker = create_new_broker(consumed_event)
    await broker.start()
    await broker.publish(queue="test-q-1", exchange="exchange")
    await consumed_event.wait()
    await broker.close()
    print("TEST END")


async def main():
    for i in range(3):
        await test_case()


if __name__ == "__main__":
    asyncio.run(main())
```

There are no documentation changes included in this PR. Please let me know if any updates are needed.

Thank you for your review.




## Type of change

- [x] Bug fix (a non-breaking change that resolves an issue)

## 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
- [ ] 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/faststream/log/logging.py b/faststream/log/logging.py
index d10852c4..6384ab76 100644
--- a/faststream/log/logging.py
+++ b/faststream/log/logging.py
@@ -57,10 +57,21 @@ def get_broker_logger(
     return logger
 
 
+def _handler_exists(logger: logging.Logger) -> bool:
+    # Check if a StreamHandler for sys.stdout already exists in the logger.
+    for handler in logger.handlers:
+        if isinstance(handler, logging.StreamHandler) and handler.stream == sys.stdout:
+            return True
+    return False
+
+
 def set_logger_fmt(
     logger: logging.Logger,
     fmt: str = "%(asctime)s %(levelname)s - %(message)s",
 ) -> None:
+    if _handler_exists(logger):
+        return
+
     handler = logging.StreamHandler(stream=sys.stdout)
 
     formatter = ColourizedFormatter(


Test patch:
diff --git a/tests/brokers/kafka/test_consume.py b/tests/brokers/kafka/test_consume.py
index c30a989d..de995162 100644
--- a/tests/brokers/kafka/test_consume.py
+++ b/tests/brokers/kafka/test_consume.py
@@ -497,7 +497,7 @@ class TestConsume(BrokerRealConsumeTestcase):
         async def handler(msg: str):
             pass
 
-        with patch.object(consume_broker, "logger", Mock()) as mock:
+        with patch.object(consume_broker, "logger", Mock(handlers=[])) as mock:
             async with self.patch_broker(consume_broker) as broker:
                 await broker.start()
                 await broker.close()
diff --git a/tests/log/test_logging.py b/tests/log/test_logging.py
new file mode 100644
index 00000000..0782b825
--- /dev/null
+++ b/tests/log/test_logging.py
@@ -0,0 +1,17 @@
+import logging
+from io import StringIO
+from unittest.mock import patch
+
+from faststream.log.logging import set_logger_fmt
+
+
+def test_duplicates_set_formatter():
+    logger = logging.getLogger(__file__)
+    logger.setLevel(logging.INFO)
+    log_output = StringIO()
+    with patch("sys.stdout", log_output):
+        set_logger_fmt(logger, fmt="%(message)s with format1")
+        set_logger_fmt(logger, fmt="%(message)s with format2")
+
+        logger.info("msg")
+        assert log_output.getvalue().strip() == "msg with format1"