Repository: rabbitmq-community/rstream (language: python)

Issue:
Handle MetadataUpdate 
MetadataUpdate is raised when there is a change in the stream topology.
As the first implementation, we can have an event, so the user can be aware of that.



Discussion / hints:

Fix patch:
diff --git a/docs/examples/check_connection_broken/consumer_handle_connections_issues_with_close.py b/docs/examples/check_connection_broken/consumer_handle_connections_issues_with_close.py
index c7c2817..46f556d 100644
--- a/docs/examples/check_connection_broken/consumer_handle_connections_issues_with_close.py
+++ b/docs/examples/check_connection_broken/consumer_handle_connections_issues_with_close.py
@@ -4,8 +4,8 @@
 from rstream import (
     AMQPMessage,
     Consumer,
-    DisconnectionErrorInfo,
     MessageContext,
+    OnClosedErrorInfo,
     amqp_decoder,
 )
 
@@ -15,7 +15,7 @@
 
 
 async def consume():
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         print(
             "connection has been closed from stream: "
             + str(disconnection_info.streams)
@@ -36,7 +36,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         vhost="/",
         username="guest",
         password="guest",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
     )
 
     loop = asyncio.get_event_loop()
diff --git a/docs/examples/check_connection_broken/consumer_handle_connections_issues_with_reconnection.py b/docs/examples/check_connection_broken/consumer_handle_connections_issues_with_reconnection.py
index fd5eb3b..53e5ed8 100644
--- a/docs/examples/check_connection_broken/consumer_handle_connections_issues_with_reconnection.py
+++ b/docs/examples/check_connection_broken/consumer_handle_connections_issues_with_reconnection.py
@@ -4,8 +4,8 @@
 from rstream import (
     AMQPMessage,
     Consumer,
-    DisconnectionErrorInfo,
     MessageContext,
+    OnClosedErrorInfo,
     amqp_decoder,
 )
 
@@ -15,7 +15,7 @@
 
 
 async def consume():
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         print(
             "connection has been closed from stream: "
             + str(disconnection_info.streams)
@@ -36,7 +36,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         vhost="/",
         username="guest",
         password="guest",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
     )
 
     loop = asyncio.get_event_loop()
diff --git a/docs/examples/check_connection_broken/producer_handle_connections_issues_with_close.py b/docs/examples/check_connection_broken/producer_handle_connections_issues_with_close.py
index 6b3bc53..338e501 100644
--- a/docs/examples/check_connection_broken/producer_handle_connections_issues_with_close.py
+++ b/docs/examples/check_connection_broken/producer_handle_connections_issues_with_close.py
@@ -1,11 +1,7 @@
 import asyncio
 import time
 
-from rstream import (
-    AMQPMessage,
-    DisconnectionErrorInfo,
-    Producer,
-)
+from rstream import AMQPMessage, OnClosedErrorInfo, Producer
 
 STREAM = "my-test-stream"
 MESSAGES = 10000000
@@ -13,7 +9,7 @@
 
 
 async def publish():
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         print(
             "connection has been closed from stream: "
             + str(disconnection_info.streams)
diff --git a/docs/examples/check_connection_broken/producer_handle_connections_issues_with_reconnect.py b/docs/examples/check_connection_broken/producer_handle_connections_issues_with_reconnect.py
index 13fd978..4582e55 100644
--- a/docs/examples/check_connection_broken/producer_handle_connections_issues_with_reconnect.py
+++ b/docs/examples/check_connection_broken/producer_handle_connections_issues_with_reconnect.py
@@ -1,11 +1,7 @@
 import asyncio
 import time
 
-from rstream import (
-    AMQPMessage,
-    DisconnectionErrorInfo,
-    Producer,
-)
+from rstream import AMQPMessage, OnClosedErrorInfo, Producer
 
 STREAM = "my-test-stream"
 MESSAGES = 10000000
@@ -13,7 +9,7 @@
 
 
 async def publish():
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         print(
             "connection has been closed from stream: "
             + str(disconnection_info.streams)
@@ -29,7 +25,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
     # producer will be closed at the end by the async context manager
     # both if connection is still alive or not
     async with Producer(
-        "localhost", username="guest", password="guest", connection_closed_handler=on_connection_closed
+        "localhost", username="guest", password="guest", on_close_handler=on_connection_closed
     ) as producer:
 
         # create a stream if it doesn't already exist
diff --git a/docs/examples/check_connection_broken/superstream_consumer_handle_connections_issues.py b/docs/examples/check_connection_broken/superstream_consumer_handle_connections_issues.py
index e42bf16..98abb8c 100644
--- a/docs/examples/check_connection_broken/superstream_consumer_handle_connections_issues.py
+++ b/docs/examples/check_connection_broken/superstream_consumer_handle_connections_issues.py
@@ -4,9 +4,9 @@
 from rstream import (
     AMQPMessage,
     ConsumerOffsetSpecification,
-    DisconnectionErrorInfo,
     MessageContext,
     OffsetType,
+    OnClosedErrorInfo,
     SuperStreamConsumer,
     amqp_decoder,
 )
@@ -25,7 +25,7 @@ async def on_message(msg: AMQPMessage, message_context: MessageContext):
 
 
 async def consume():
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         print(
             "connection has been closed from stream: "
             + str(disconnection_info.streams)
@@ -46,7 +46,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         username="guest",
         password="guest",
         super_stream="invoices",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
     )
 
     loop = asyncio.get_event_loop()
diff --git a/docs/examples/check_connection_broken/superstream_consumer_handle_reconnection.py b/docs/examples/check_connection_broken/superstream_consumer_handle_reconnection.py
index c2dc70b..b918038 100644
--- a/docs/examples/check_connection_broken/superstream_consumer_handle_reconnection.py
+++ b/docs/examples/check_connection_broken/superstream_consumer_handle_reconnection.py
@@ -4,9 +4,9 @@
 from rstream import (
     AMQPMessage,
     ConsumerOffsetSpecification,
-    DisconnectionErrorInfo,
     MessageContext,
     OffsetType,
+    OnClosedErrorInfo,
     SuperStreamConsumer,
     amqp_decoder,
 )
@@ -25,7 +25,7 @@ async def on_message(msg: AMQPMessage, message_context: MessageContext):
 
 
 async def consume():
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         print(
             "connection has been closed from stream: "
             + str(disconnection_info.streams)
@@ -45,7 +45,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         username="guest",
         password="guest",
         super_stream="invoices",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
     )
 
     loop = asyncio.get_event_loop()
diff --git a/docs/examples/check_connection_broken/superstream_producer_handle_connections_issues.py b/docs/examples/check_connection_broken/superstream_producer_handle_connections_issues.py
index 101ed3a..b5a0694 100644
--- a/docs/examples/check_connection_broken/superstream_producer_handle_connections_issues.py
+++ b/docs/examples/check_connection_broken/superstream_producer_handle_connections_issues.py
@@ -3,7 +3,7 @@
 
 from rstream import (
     AMQPMessage,
-    DisconnectionErrorInfo,
+    OnClosedErrorInfo,
     RouteType,
     SuperStreamProducer,
 )
@@ -19,7 +19,7 @@ async def publish():
     async def routing_extractor(message: AMQPMessage) -> str:
         return message.application_properties["id"]
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
 
         print(
             "connection has been closed from stream: "
@@ -39,7 +39,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         password="guest",
         routing_extractor=routing_extractor,
         routing=RouteType.Hash,
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         super_stream=SUPER_STREAM,
     ) as super_stream_producer:
 
diff --git a/docs/examples/check_connection_broken/superstream_producer_handle_reconnection.py b/docs/examples/check_connection_broken/superstream_producer_handle_reconnection.py
index 0d47965..b89907d 100644
--- a/docs/examples/check_connection_broken/superstream_producer_handle_reconnection.py
+++ b/docs/examples/check_connection_broken/superstream_producer_handle_reconnection.py
@@ -3,7 +3,7 @@
 
 from rstream import (
     AMQPMessage,
-    DisconnectionErrorInfo,
+    OnClosedErrorInfo,
     RouteType,
     SuperStreamProducer,
 )
@@ -19,7 +19,7 @@ async def publish():
     async def routing_extractor(message: AMQPMessage) -> str:
         return message.application_properties["id"]
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
 
         print(
             "connection has been closed from stream: "
@@ -41,7 +41,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         password="guest",
         routing_extractor=routing_extractor,
         routing=RouteType.Hash,
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         super_stream=SUPER_STREAM,
     ) as super_stream_producer:
 
diff --git a/docs/examples/metadata_update/README.md b/docs/examples/metadata_update/README.md
new file mode 100644
index 0000000..cd59b02
--- /dev/null
+++ b/docs/examples/metadata_update/README.md
@@ -0,0 +1,18 @@
+Simple scenarios for Metadata update management
+---
+
+If the streams topology changes (ex:Stream deleted or add/remove follower), the client receives an MetadataUpdate event.
+
+The client is notified by a callback (similar to what happens for connection_broken scenarios)
+
+After this the client can try to reconnect.
+
+Here you can find examples for producers and consumers.
+
+You can start the producer and the consumer and then force the server to change the topology of the stream for example with this command:
+
+rabbitmq-streams delete_replica my-test-stream rabbit@rabbitmqcluster-sample-server-0.rabbitmqcluster-sample-nodes.rabbitmq-system
+
+The examples need to run in a minimal 3 nodes RabbitMQ cluster.
+
+There is also an example where you can manage both Metadata Update and disconnections
\ No newline at end of file
diff --git a/docs/examples/metadata_update/consumer_metadata_disconnection.py b/docs/examples/metadata_update/consumer_metadata_disconnection.py
new file mode 100644
index 0000000..572ef16
--- /dev/null
+++ b/docs/examples/metadata_update/consumer_metadata_disconnection.py
@@ -0,0 +1,84 @@
+import asyncio
+import signal
+
+from rstream import (
+    AMQPMessage,
+    Consumer,
+    MessageContext,
+    OnClosedErrorInfo,
+    amqp_decoder,
+)
+
+STREAM = "my-test-stream"
+COUNT = 0
+consumer_closed = False
+
+
+async def print_count_values():
+
+    global COUNT
+
+    while True:
+        await asyncio.sleep(5)
+
+        print("NUMBER OF CONSUMED MESSAGES:         " + str(COUNT))
+
+
+async def consume():
+    asyncio.create_task(print_count_values())
+
+    async def connect() -> Consumer:
+        async def on_metadata_update(on_closed_info: OnClosedErrorInfo) -> None:
+            if on_closed_info.reason == "MetaData Update":
+                print(
+                    "metadata changed for stream : "
+                    + str(on_closed_info.streams[0])
+                    + " with code: "
+                    + on_closed_info.reason
+                )
+                await asyncio.sleep(2)
+                # reconnect just if the stream exists
+                if await consumer.stream_exists(on_closed_info.streams[0]):
+                    await consumer.reconnect_stream(on_closed_info.streams[0])
+                else:
+                    await consumer.close()
+
+            else:
+                print(
+                    "connection has been closed from stream: "
+                    + str(on_closed_info.streams)
+                    + " for reason: "
+                    + str(on_closed_info.reason)
+                )
+                await asyncio.sleep(2)
+                # reconnect just if the stream exists
+                for stream in on_closed_info.streams:
+                    await consumer.reconnect_stream(stream)
+
+        consumer = Consumer(
+            host="localhost",
+            port=5552,
+            vhost="/",
+            username="guest",
+            password="guest",
+            on_close_handler=on_metadata_update,
+        )
+
+        return consumer
+
+    loop = asyncio.get_event_loop()
+    loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))
+
+    async def on_message(msg: AMQPMessage, message_context: MessageContext):
+        stream = message_context.consumer.get_stream(message_context.subscriber_name)
+        offset = message_context.offset
+        global COUNT
+        COUNT = COUNT + 1
+
+    consumer = await connect()
+    await consumer.start()
+    await consumer.subscribe(stream=STREAM, callback=on_message, decoder=amqp_decoder)
+    await consumer.run()
+
+
+asyncio.run(consume())
diff --git a/docs/examples/metadata_update/consumer_metadata_update.py b/docs/examples/metadata_update/consumer_metadata_update.py
new file mode 100644
index 0000000..eb95609
--- /dev/null
+++ b/docs/examples/metadata_update/consumer_metadata_update.py
@@ -0,0 +1,63 @@
+import asyncio
+import signal
+
+from rstream import (
+    AMQPMessage,
+    Consumer,
+    MessageContext,
+    OnClosedErrorInfo,
+    amqp_decoder,
+)
+
+STREAM = "my-test-stream"
+COUNT = 0
+consumer_closed = False
+
+
+async def consume():
+    async def connect() -> Consumer:
+        async def on_metadata_update(metadata_update_info: OnClosedErrorInfo) -> None:
+            print(
+                "metadata changed for stream : "
+                + str(metadata_update_info.streams[0])
+                + " with code: "
+                + metadata_update_info.reason
+            )
+
+            await asyncio.sleep(2)
+            # reconnect just if the stream exists
+            if await consumer.stream_exists(metadata_update_info.streams[0]):
+                await consumer.reconnect_stream(metadata_update_info.streams[0])
+            else:
+                await consumer.close()
+
+        consumer = Consumer(
+            host="localhost",
+            port=5552,
+            vhost="/",
+            username="guest",
+            password="guest",
+            on_close_handler=on_metadata_update,
+        )
+
+        return consumer
+
+    loop = asyncio.get_event_loop()
+    loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))
+
+    async def on_message(msg: AMQPMessage, message_context: MessageContext):
+        stream = message_context.consumer.get_stream(message_context.subscriber_name)
+        offset = message_context.offset
+        global COUNT
+        COUNT = COUNT + 1
+        if COUNT % 100000 == 0:
+            print("Got message: {} from stream {}, offset {}".format(msg, stream, offset))
+            print("consumed 1 million messages")
+
+    consumer = await connect()
+    await consumer.start()
+    await consumer.subscribe(stream=STREAM, callback=on_message, decoder=amqp_decoder)
+    await consumer.run()
+
+
+asyncio.run(consume())
diff --git a/docs/examples/metadata_update/producer_metadata_disconnection.py b/docs/examples/metadata_update/producer_metadata_disconnection.py
new file mode 100644
index 0000000..18a304a
--- /dev/null
+++ b/docs/examples/metadata_update/producer_metadata_disconnection.py
@@ -0,0 +1,112 @@
+import asyncio
+import time
+
+from rstream import (
+    AMQPMessage,
+    ConfirmationStatus,
+    OnClosedErrorInfo,
+    Producer,
+)
+
+STREAM = "my-test-stream"
+MESSAGES = 10000000
+producer_closed = False
+COUNT = 0
+CONFIRMED = 0
+
+
+async def _on_publish_confirm_client(confirmation: ConfirmationStatus) -> None:
+    global CONFIRMED
+    if confirmation.is_confirmed:
+        CONFIRMED = CONFIRMED + 1
+
+
+async def print_count_values():
+
+    global COUNT
+    global CONFIRMED
+
+    while True:
+        await asyncio.sleep(5)
+
+        print("NUMBER OF PRODUCED MESSAGES:         " + str(COUNT))
+        print("NUMBER OF CONFIRMED MESSAGES:        " + str(CONFIRMED))
+
+
+async def publish():
+    asyncio.create_task(print_count_values())
+
+    async def on_metadata_update(on_closed_info: OnClosedErrorInfo) -> None:
+
+        global producer_closed
+        producer_closed = True
+
+        if on_closed_info.reason == "MetaData Update":
+            print(
+                "metadata changed for stream : "
+                + str(on_closed_info.streams[0])
+                + " with code: "
+                + on_closed_info.reason
+            )
+
+            await asyncio.sleep(2)
+            if await producer.stream_exists(on_closed_info.streams[0]):
+                await producer.reconnect_stream(on_closed_info.streams[0])
+            else:
+                await producer.close()
+
+        else:
+            print(
+                "connection has been closed from stream: "
+                + str(on_closed_info.streams)
+                + " for reason: "
+                + str(on_closed_info.reason)
+            )
+
+            for stream in on_closed_info.streams:
+                await asyncio.sleep(2)
+                await producer.reconnect_stream(stream)
+
+    print("creating Producer")
+    # producer will be closed at the end by the async context manager
+    # both if connection is still alive or not
+    async with Producer(
+        "localhost",
+        username="guest",
+        password="guest",
+        on_close_handler=on_metadata_update,
+    ) as producer:
+
+        # create a stream if it doesn't already exist
+        await producer.create_stream(STREAM, exists_ok=True)
+
+        # sending a million of messages in AMQP format
+        start_time = time.perf_counter()
+
+        print("Sending MESSAGES")
+        global COUNT
+        for i in range(MESSAGES):
+            amqp_message = AMQPMessage(
+                body="hello: {}".format(i),
+            )
+            # send is asynchronous
+            global producer_closed
+            if producer_closed is False:
+                try:
+                    await producer.send(
+                        stream=STREAM, message=amqp_message, on_publish_confirm=_on_publish_confirm_client
+                    )
+                except Exception as e:
+                    await asyncio.sleep(3)
+                    continue
+            else:
+                producer_closed = False
+                continue
+
+            COUNT = COUNT + 1
+
+    end_time = time.perf_counter()
+    print(f"Sent {MESSAGES} messages in {end_time - start_time:0.4f} seconds")
+
+
+asyncio.run(publish())
diff --git a/docs/examples/metadata_update/producer_metadata_update.py b/docs/examples/metadata_update/producer_metadata_update.py
new file mode 100644
index 0000000..1ea2810
--- /dev/null
+++ b/docs/examples/metadata_update/producer_metadata_update.py
@@ -0,0 +1,80 @@
+import asyncio
+import time
+
+from rstream import AMQPMessage, OnClosedErrorInfo, Producer
+
+STREAM = "my-test-stream"
+MESSAGES = 10000000
+producer_closed = False
+
+
+async def publish():
+    async def on_metadata_update(on_closed_info: OnClosedErrorInfo) -> None:
+
+        if on_closed_info.reason == "MetaData Update":
+            print(
+                "metadata changed for stream : "
+                + str(on_closed_info.streams[0])
+                + " with code: "
+                + on_closed_info.reason
+            )
+
+        else:
+            print(
+                "connection has been closed from stream: "
+                + str(on_closed_info.streams)
+                + " for reason: "
+                + str(on_closed_info.reason)
+            )
+
+        global producer_closed
+        producer_closed = True
+        await asyncio.sleep(2)
+        # reconnect if the stream is still existing
+        if await producer.stream_exists(on_closed_info.streams[0]):
+            await producer.reconnect_stream(on_closed_info.streams[0])
+        else:
+            await producer.close()
+
+    print("creating Producer")
+    # producer will be closed at the end by the async context manager
+    # both if connection is still alive or not
+    async with Producer(
+        "localhost",
+        username="guest",
+        password="guest",
+        on_close_handler=on_metadata_update,
+    ) as producer:
+
+        # create a stream if it doesn't already exist
+        await producer.create_stream(STREAM, exists_ok=True)
+
+        # sending a million of messages in AMQP format
+        start_time = time.perf_counter()
+
+        print("Sending MESSAGES")
+        for i in range(MESSAGES):
+            amqp_message = AMQPMessage(
+                body="hello: {}".format(i),
+            )
+            # send is asynchronous
+            global producer_closed
+            if producer_closed is False:
+                try:
+                    await producer.send(stream=STREAM, message=amqp_message)
+                except Exception as e:
+                    # gives time to reconnect_stream to reconnect
+                    await asyncio.sleep(3)
+                    continue
+            else:
+                producer_closed = False
+                break
+
+            if i % 100000 == 0:
+                print("sent 10000 messages")
+
+    end_time = time.perf_counter()
+    print(f"Sent {MESSAGES} messages in {end_time - start_time:0.4f} seconds")
+
+
+asyncio.run(publish())
diff --git a/docs/examples/metadata_update/super_stream_consumer_metadata_update.py b/docs/examples/metadata_update/super_stream_consumer_metadata_update.py
new file mode 100644
index 0000000..278ed99
--- /dev/null
+++ b/docs/examples/metadata_update/super_stream_consumer_metadata_update.py
@@ -0,0 +1,76 @@
+import asyncio
+import signal
+
+from rstream import (
+    AMQPMessage,
+    ConsumerOffsetSpecification,
+    MessageContext,
+    OffsetType,
+    OnClosedErrorInfo,
+    SuperStreamConsumer,
+    amqp_decoder,
+)
+
+count = 0
+connection_is_closed = False
+
+
+async def on_message(msg: AMQPMessage, message_context: MessageContext):
+    global count
+    count += 1
+    if (count % 100000) == 0:
+        stream = await message_context.consumer.stream(message_context.subscriber_name)
+        offset = message_context.offset
+        print("Received message: {} from stream: {} - message offset: {}".format(msg, stream, offset))
+
+
+async def consume():
+    async def on_metadata_update(on_closed_info: OnClosedErrorInfo) -> None:
+
+        if on_closed_info.reason == "MetaData Update":
+            print(
+                "metadata changed for stream : "
+                + str(on_closed_info.streams[0])
+                + " with code: "
+                + on_closed_info.reason
+            )
+
+            for stream in on_closed_info.streams:
+                await asyncio.sleep(2)
+                if await consumer.stream_exists(on_closed_info.streams[0]):
+                    await consumer.reconnect_stream(stream)
+
+        else:
+            print(
+                "connection has been closed from stream: "
+                + str(on_closed_info.streams)
+                + " for reason: "
+                + str(on_closed_info.reason)
+            )
+
+            for stream in on_closed_info.streams:
+                await asyncio.sleep(2)
+                await consumer.reconnect_stream(stream)
+
+    consumer = SuperStreamConsumer(
+        host="34.89.82.143",
+        port=5552,
+        vhost="/",
+        username="XXXXXXXXXXXX",
+        password="XXXXXXXXXXXX",
+        super_stream="invoices",
+        load_balancer_mode=True,
+        on_close_handler=on_metadata_update,
+    )
+
+    loop = asyncio.get_event_loop()
+    loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))
+    offset_specification = ConsumerOffsetSpecification(OffsetType.FIRST, None)
+    await consumer.start()
+    await consumer.subscribe(
+        callback=on_message, decoder=amqp_decoder, offset_specification=offset_specification
+    )
+    await consumer.run()
+
+
+asyncio.run(consume())
diff --git a/docs/examples/metadata_update/super_stream_producer_metadata_update.py b/docs/examples/metadata_update/super_stream_producer_metadata_update.py
new file mode 100644
index 0000000..05fec67
--- /dev/null
+++ b/docs/examples/metadata_update/super_stream_producer_metadata_update.py
@@ -0,0 +1,88 @@
+import asyncio
+import time
+
+from rstream import (
+    AMQPMessage,
+    OnClosedErrorInfo,
+    RouteType,
+    SuperStreamProducer,
+)
+
+SUPER_STREAM = "invoices"
+MESSAGES = 10000000
+producer_closed = False
+
+# this value will be hashed using mumh3 hashing algorithm to decide the partition resolution for the message
+async def routing_extractor(message: AMQPMessage) -> str:
+    return message.application_properties["id"]
+
+
+async def publish():
+    async def on_metadata_update(on_closed_info: OnClosedErrorInfo) -> None:
+
+        global producer_closed
+        producer_closed = True
+
+        if on_closed_info.reason == "MetaData Update":
+            print(
+                "metadata changed for stream : "
+                + str(on_closed_info.streams[0])
+                + " with code: "
+                + on_closed_info.reason
+            )
+            await asyncio.sleep(2)
+            # reconnect just if the partition exists
+            if await super_stream_producer.stream_exists((on_closed_info.streams[0])):
+                await super_stream_producer.reconnect_stream(on_closed_info.streams[0])
+
+        else:
+            print(
+                "connection has been closed from stream: "
+                + str(on_closed_info.streams)
+                + " for reason: "
+                + str(on_closed_info.reason)
+            )
+            await asyncio.sleep(2)
+            # reconnect just if the partition exists
+            for stream in on_closed_info.streams:
+                await super_stream_producer.reconnect_stream(stream)
+
+    # SuperStreamProducer wraps a Producer
+    async with SuperStreamProducer(
+        "localhost",
+        username="guest",
+        password="guest",
+        routing_extractor=routing_extractor,
+        routing=RouteType.Hash,
+        super_stream=SUPER_STREAM,
+        on_close_handler=on_metadata_update,
+    ) as super_stream_producer:
+        # Sending a million messages
+
+        start_time = time.perf_counter()
+        for i in range(MESSAGES):
+            amqp_message = AMQPMessage(
+                body="hello: {}".format(i),
+                application_properties={"id": "{}".format(i)},
+            )
+            global producer_closed
+            if producer_closed is False:
+                try:
+                    await super_stream_producer.send(amqp_message)
+                except Exception as e:
+                    # give some time to the reconnect_stream to reconnect
+                    await asyncio.sleep(3)
+                    continue
+            else:
+                producer_closed = False
+                continue
+            if i % 100000 == 0:
+                print(f"Published {i} messages to super stream: {SUPER_STREAM}")
+
+        end_time = time.perf_counter()
+        print(
+            f"Sent {MESSAGES} messages in {end_time - start_time:0.4f} seconds to super stream: {SUPER_STREAM}"
+        )
+
+
+asyncio.run(publish())
diff --git a/rstream/__init__.py b/rstream/__init__.py
index 5eed7d0..bafc7dc 100644
--- a/rstream/__init__.py
+++ b/rstream/__init__.py
@@ -3,10 +3,7 @@
 
 from importlib import metadata
 
-from .utils import (
-    DisconnectionErrorInfo,
-    FilterConfiguration,
-)
+from .utils import FilterConfiguration, OnClosedErrorInfo
 
 try:
     __version__ = metadata.version(__package__)
@@ -65,7 +62,7 @@
     "StreamDoesNotExist",
     "OffsetSpecification",
     "EventContext",
-    "DisconnectionErrorInfo",
+    "OnClosedErrorInfo",
     "SlasMechanism",
     "FilterConfiguration",
 ]
diff --git a/rstream/client.py b/rstream/client.py
index d6b3653..d8a49a6 100644
--- a/rstream/client.py
+++ b/rstream/client.py
@@ -34,7 +34,7 @@
 from .connection import Connection, ConnectionClosed
 from .constants import SlasMechanism
 from .schema import OffsetSpecification
-from .utils import DisconnectionErrorInfo
+from .utils import OnClosedErrorInfo
 
 FT = TypeVar("FT", bound=schema.Frame)
 HT = Annotated[
@@ -69,7 +69,7 @@ def __init__(
         frame_max: int,
         heartbeat: int,
         connection_name: Optional[str] = "",
-        connection_closed_handler: Optional[CB[DisconnectionErrorInfo]] = None,
+        connection_closed_handler: Optional[CB[OnClosedErrorInfo]] = None,
         sasl_configuration_mechanism: SlasMechanism = SlasMechanism.MechanismPlain,
     ):
         self.host = host
@@ -268,7 +268,7 @@ async def _listener(self) -> None:
             if self._connection_closed_handler is not None:
                 # don't raise for locator connections without streams
                 if len(self._streams) > 0:
-                    connection_error_info = DisconnectionErrorInfo("Connection Closed", self._streams)
+                    connection_error_info = OnClosedErrorInfo("Connection Closed", self._streams)
                     result = self._connection_closed_handler(connection_error_info)
                     if result is not None and inspect.isawaitable(result):
                         await result
@@ -663,7 +663,7 @@ async def get(
         self,
         connection_name: Optional[str],
         addr: Optional[Addr] = None,
-        connection_closed_handler: Optional[CB[DisconnectionErrorInfo]] = None,
+        connection_closed_handler: Optional[CB[OnClosedErrorInfo]] = None,
         stream: Optional[str] = None,
         sasl_configuration_mechanism: SlasMechanism = SlasMechanism.MechanismPlain,
     ) -> Client:
@@ -702,7 +702,7 @@ async def _resolve_broker(
         self,
         connection_name: Optional[str],
         addr: Addr,
-        connection_closed_handler: Optional[CB[DisconnectionErrorInfo]] = None,
+        connection_closed_handler: Optional[CB[OnClosedErrorInfo]] = None,
         sasl_configuration_mechanism: SlasMechanism = SlasMechanism.MechanismPlain,
     ) -> Client:
         desired_host, desired_port = addr.host, str(addr.port)
@@ -738,7 +738,7 @@ async def new(
         self,
         connection_name: Optional[str],
         addr: Addr,
-        connection_closed_handler: Optional[CB[DisconnectionErrorInfo]] = None,
+        connection_closed_handler: Optional[CB[OnClosedErrorInfo]] = None,
         sasl_configuration_mechanism: SlasMechanism = SlasMechanism.MechanismPlain,
     ) -> Client:
         host, port = addr
diff --git a/rstream/consumer.py b/rstream/consumer.py
index 26a3219..efb9594 100644
--- a/rstream/consumer.py
+++ b/rstream/consumer.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 import logging
 import random
 import ssl
@@ -30,10 +31,7 @@
     SlasMechanism,
 )
 from .schema import OffsetSpecification
-from .utils import (
-    DisconnectionErrorInfo,
-    FilterConfiguration,
-)
+from .utils import FilterConfiguration, OnClosedErrorInfo
 
 MT = TypeVar("MT")
 CB = Annotated[Callable[[MT, Any], Union[None, Awaitable[None]]], "Message callback type"]
@@ -82,7 +80,7 @@ def __init__(
         heartbeat: int = 60,
         load_balancer_mode: bool = False,
         max_retries: int = 20,
-        connection_closed_handler: Optional[CB_CONN[DisconnectionErrorInfo]] = None,
+        on_close_handler: Optional[CB_CONN[OnClosedErrorInfo]] = None,
         connection_name: str = None,
         sasl_configuration_mechanism: SlasMechanism = SlasMechanism.MechanismPlain,
     ):
@@ -105,7 +103,7 @@ def __init__(
         self._subscribers: dict[str, _Subscriber] = {}
         self._stop_event = asyncio.Event()
         self._lock = asyncio.Lock()
-        self._connection_closed_handler = connection_closed_handler
+        self._on_close_handler = on_close_handler
         self._connection_name = connection_name
         self._sasl_configuration_mechanism = sasl_configuration_mechanism
         if self._connection_name is None:
@@ -126,7 +124,7 @@ async def __aexit__(self, *_: Any) -> None:
 
     async def start(self) -> None:
         self._default_client = await self._pool.get(
-            connection_closed_handler=self._connection_closed_handler,
+            connection_closed_handler=self._on_close_handler,
             connection_name=self._connection_name,
             sasl_configuration_mechanism=self._sasl_configuration_mechanism,
         )
@@ -154,14 +152,14 @@ async def _get_or_create_client(self, stream: str) -> Client:
         if stream not in self._clients:
             if self._default_client is None:
                 self._default_client = await self._pool.get(
-                    connection_closed_handler=self._connection_closed_handler,
+                    connection_closed_handler=self._on_close_handler,
                     connection_name=self._connection_name,
                 )
             leader, replicas = await (await self.default_client).query_leader_and_replicas(stream)
             broker = random.choice(replicas) if replicas else leader
             self._clients[stream] = await self._pool.get(
                 addr=Addr(broker.host, broker.port),
-                connection_closed_handler=self._connection_closed_handler,
+                connection_closed_handler=self._on_close_handler,
                 connection_name=self._connection_name,
                 stream=stream,
             )
@@ -236,6 +234,12 @@ async def subscribe(
             name=subscriber.reference,
         )
 
+        subscriber.client.add_handler(
+            schema.MetadataUpdate,
+            partial(self._on_metadata_update),
+            name=subscriber.reference,
+        )
+
         # to handle single-active-consumer
         if properties is not None:
             if "single-active-consumer" in properties:
@@ -284,6 +288,10 @@ async def unsubscribe(self, subscriber_name: str) -> None:
             schema.Deliver,
             name=subscriber.reference,
         )
+        subscriber.client.remove_handler(
+            schema.MetadataUpdate,
+            name=subscriber.reference,
+        )
         try:
             await asyncio.wait_for(subscriber.client.unsubscribe(subscriber.subscription_id), 5)
         except asyncio.TimeoutError:
@@ -356,6 +364,14 @@ async def _on_deliver(
             if maybe_coro is not None:
                 await maybe_coro
 
+    async def _on_metadata_update(self, frame: schema.MetadataUpdate) -> None:
+
+        if self._on_close_handler is not None:
+            metadata_update_info = OnClosedErrorInfo("MetaData Update", [frame.metadata_info.stream])
+            result = self._on_close_handler(metadata_update_info)
+            if result is not None and inspect.isawaitable(result):
+                await result
+
     async def _on_consumer_update_query_response(
         self,
         frame: schema.ConsumerUpdateResponse,
@@ -407,8 +423,10 @@ async def delete_stream(self, stream: str, missing_ok: bool = False) -> None:
             finally:
                 await self._close_locator_connection()
 
-    async def stream_exists(self, stream: str) -> bool:
+    async def stream_exists(self, stream: str, on_close_event: bool = False) -> bool:
         async with self._lock:
+            if on_close_event:
+                self._default_client = None
             stream_exists = await (await self.default_client).stream_exists(stream)
             await self._close_locator_connection()
 
@@ -473,7 +491,7 @@ async def _check_if_filtering_is_supported(self) -> None:
 
     async def _create_locator_connection(self) -> Client:
         return await self._pool.get(
-            connection_closed_handler=self._connection_closed_handler,
+            connection_closed_handler=self._on_close_handler,
             connection_name=self._connection_name,
             sasl_configuration_mechanism=self._sasl_configuration_mechanism,
         )
diff --git a/rstream/producer.py b/rstream/producer.py
index 955e36a..4895706 100644
--- a/rstream/producer.py
+++ b/rstream/producer.py
@@ -31,7 +31,7 @@
     ICompressionCodec,
 )
 from .constants import Key, SlasMechanism
-from .utils import DisconnectionErrorInfo, RawMessage
+from .utils import OnClosedErrorInfo, RawMessage
 
 MessageT = TypeVar("MessageT", _MessageProtocol, bytes)
 message_v1_v2 = T = TypeVar("message_v1_v2")
@@ -83,7 +83,7 @@ def __init__(
         default_batch_publishing_delay: float = 0.2,
         default_context_switch_value: int = 1000,
         connection_name: str = None,
-        connection_closed_handler: Optional[CB[DisconnectionErrorInfo]] = None,
+        on_close_handler: Optional[CB[OnClosedErrorInfo]] = None,
         sasl_configuration_mechanism: SlasMechanism = SlasMechanism.MechanismPlain,
         filter_value_extractor: Optional[CB_F[Any]] = None,
     ):
@@ -115,7 +115,7 @@ def __init__(
         self._default_batch_publishing_delay = default_batch_publishing_delay
         self._default_context_switch_counter = 0
         self._default_context_switch_value = default_context_switch_value
-        self._connection_closed_handler = connection_closed_handler
+        self._on_close_handler = on_close_handler
         self._sasl_configuration_mechanism = sasl_configuration_mechanism
         self._close_called = False
         self._connection_name = connection_name
@@ -140,7 +140,7 @@ async def __aexit__(self, *_: Any) -> None:
     async def start(self) -> None:
         self._close_called = False
         self._default_client = await self._pool.get(
-            connection_closed_handler=self._connection_closed_handler,
+            connection_closed_handler=self._on_close_handler,
             connection_name=self._connection_name,
             sasl_configuration_mechanism=self._sasl_configuration_mechanism,
         )
@@ -183,6 +183,7 @@ async def close(self) -> None:
                     logger.debug("exception in delete_publisher in Producer.close:", exc)
             publisher.client.remove_handler(schema.PublishConfirm, publisher.reference)
             publisher.client.remove_handler(schema.PublishError, publisher.reference)
+            publisher.client.remove_handler(schema.MetadataUpdate, publisher.reference)
 
         self._publishers.clear()
 
@@ -195,14 +196,14 @@ async def _get_or_create_client(self, stream: str) -> Client:
         if stream not in self._clients:
             if self._default_client is None:
                 self._default_client = await self._pool.get(
-                    connection_closed_handler=self._connection_closed_handler,
+                    connection_closed_handler=self._on_close_handler,
                     connection_name=self._connection_name,
                 )
             leader, _ = await (await self.default_client).query_leader_and_replicas(stream)
             self._clients[stream] = await self._pool.get(
                 connection_name=self._connection_name,
                 addr=Addr(leader.host, leader.port),
-                connection_closed_handler=self._connection_closed_handler,
+                connection_closed_handler=self._on_close_handler,
                 stream=stream,
                 sasl_configuration_mechanism=self._sasl_configuration_mechanism,
             )
@@ -255,6 +256,11 @@ async def _get_or_create_publisher(
             partial(self._on_publish_error, publisher=publisher),
             name=publisher.reference,
         )
+        client.add_handler(
+            schema.MetadataUpdate,
+            partial(self._on_metadata_update),
+            name=publisher.reference,
+        )
 
         return publisher
 
@@ -627,6 +633,14 @@ async def _on_publish_error(self, frame: schema.PublishError, publisher: _Publis
                     if isinstance(confirmation, asyncio.Future):
                         confirmation.set_exception(exc)
 
+    async def _on_metadata_update(self, frame: schema.MetadataUpdate) -> None:
+
+        if self._on_close_handler is not None:
+            metadata_update_info = OnClosedErrorInfo("MetaData Update", [frame.metadata_info.stream])
+            result = self._on_close_handler(metadata_update_info)
+            if result is not None and inspect.isawaitable(result):
+                await result
+
     async def create_stream(
         self,
         stream: str,
@@ -649,6 +663,7 @@ async def delete_stream(self, stream: str, missing_ok: bool = False) -> None:
             await publisher.client.delete_publisher(publisher.id)
             publisher.client.remove_handler(schema.PublishConfirm, publisher.reference)
             publisher.client.remove_handler(schema.PublishError, publisher.reference)
+            publisher.client.remove_handler(schema.MetadataUpdate, publisher.reference)
             del self._publishers[stream]
 
         async with self._lock:
@@ -660,8 +675,11 @@ async def delete_stream(self, stream: str, missing_ok: bool = False) -> None:
             finally:
                 await self._close_locator_connection()
 
-    async def stream_exists(self, stream: str) -> bool:
+    async def stream_exists(self, stream: str, on_close_event: bool = False) -> bool:
+
         async with self._lock:
+            if on_close_event:
+                self._default_client = None
             stream_exist = await (await self.default_client).stream_exists(stream)
             await self._close_locator_connection()
         return stream_exist
@@ -708,7 +726,7 @@ async def _check_if_filtering_is_supported(self) -> None:
 
     async def _create_locator_connection(self) -> Client:
         return await self._pool.get(
-            connection_closed_handler=self._connection_closed_handler,
+            connection_closed_handler=self._on_close_handler,
             connection_name=self._connection_name,
             sasl_configuration_mechanism=self._sasl_configuration_mechanism,
         )
diff --git a/rstream/superstream_consumer.py b/rstream/superstream_consumer.py
index c06385a..85f1827 100644
--- a/rstream/superstream_consumer.py
+++ b/rstream/superstream_consumer.py
@@ -25,10 +25,7 @@
 )
 from .consumer import Consumer, EventContext, MessageContext
 from .superstream import DefaultSuperstreamMetadata
-from .utils import (
-    DisconnectionErrorInfo,
-    FilterConfiguration,
-)
+from .utils import FilterConfiguration, OnClosedErrorInfo
 
 MT = TypeVar("MT")
 CB = Annotated[Callable[[MT], Union[None, Awaitable[None]]], "Message callback type"]
@@ -50,7 +47,7 @@ def __init__(
         max_retries: int = 20,
         super_stream: str,
         connection_name: str = None,
-        connection_closed_handler: Optional[CB[DisconnectionErrorInfo]] = None,
+        on_close_handler: Optional[CB[OnClosedErrorInfo]] = None,
     ):
         self._pool = ClientPool(
             host,
@@ -81,7 +78,7 @@ def __init__(
         self._consumers: dict[str, Consumer] = {}
         self._stop_event = asyncio.Event()
         self._subscribers: dict[str, str] = defaultdict(str)
-        self._connection_closed_handler = connection_closed_handler
+        self._on_close_handler = on_close_handler
         self._connection_name = connection_name
         if self._connection_name is None:
             self._connection_name = "rstream-consumer"
@@ -124,7 +121,7 @@ async def _get_or_create_client(self, stream: str) -> Client:
             broker = random.choice(replicas) if replicas else leader
             self._clients[stream] = await self._pool.get(
                 addr=Addr(broker.host, broker.port),
-                connection_closed_handler=self._connection_closed_handler,
+                connection_closed_handler=self._on_close_handler,
                 connection_name=self._connection_name,
                 stream=stream,
             )
@@ -153,7 +150,7 @@ async def subscribe(
 
         if self._default_client is None or self._default_client.is_connection_alive() is False:
             self._default_client = await self._pool.get(
-                connection_closed_handler=self._connection_closed_handler, connection_name="rstream-locator"
+                connection_closed_handler=self._on_close_handler, connection_name="rstream-locator"
             )
 
         self._super_stream_metadata = DefaultSuperstreamMetadata(self.super_stream, self.default_client)
@@ -193,7 +190,7 @@ async def _create_consumer(self) -> Consumer:
             heartbeat=self.heartbeat,
             load_balancer_mode=self.load_balancer_mode,
             max_retries=self.max_retries,
-            connection_closed_handler=self._connection_closed_handler,
+            on_close_handler=self._on_close_handler,
             connection_name=self._connection_name,
         )
 
@@ -213,3 +210,7 @@ async def unsubscribe(self) -> None:
 
     async def reconnect_stream(self, stream: str, offset: Optional[int] = None) -> None:
         await self._consumers[stream].reconnect_stream(stream, offset)
+
+    async def stream_exists(self, stream: str) -> bool:
+        stream_exist = await self._consumers[stream].stream_exists(stream)
+        return stream_exist
diff --git a/rstream/superstream_producer.py b/rstream/superstream_producer.py
index 9155c8a..f895660 100644
--- a/rstream/superstream_producer.py
+++ b/rstream/superstream_producer.py
@@ -21,7 +21,7 @@
     RoutingKeyRoutingStrategy,
     RoutingStrategy,
 )
-from .utils import DisconnectionErrorInfo
+from .utils import OnClosedErrorInfo
 
 MT = TypeVar("MT")
 CB = Annotated[Callable[[MT], Awaitable[Any]], "Message callback type"]
@@ -53,7 +53,7 @@ def __init__(
         load_balancer_mode: bool = False,
         max_retries: int = 20,
         default_batch_publishing_delay: float = 0.2,
-        connection_closed_handler: Optional[CB[DisconnectionErrorInfo]] = None,
+        on_close_handler: Optional[CB[OnClosedErrorInfo]] = None,
         connection_name: str = None,
         filter_value_extractor: Optional[CB_F[Any]] = None,
     ):
@@ -86,7 +86,7 @@ def __init__(
         self._default_client: Optional[Client] = None
         self._producer: Producer | None = None
         self._routing_strategy: RoutingStrategy
-        self._connection_closed_handler = connection_closed_handler
+        self._on_close_handler = on_close_handler
         self._connection_name = connection_name
         if self._connection_name is None:
             self._connection_name = "rstream-producer"
@@ -105,7 +105,7 @@ async def _get_producer(self) -> Producer:
                 heartbeat=self.heartbeat,
                 load_balancer_mode=self.load_balancer_mode,
                 default_batch_publishing_delay=self.default_batch_publishing_delay,
-                connection_closed_handler=self._connection_closed_handler,
+                on_close_handler=self._on_close_handler,
                 connection_name=self._connection_name,
                 filter_value_extractor=self._filter_value_extractor,
             )
@@ -140,7 +140,7 @@ async def __aexit__(self, *_: Any) -> None:
 
     async def start(self) -> None:
         self._default_client = await self._pool.get(
-            connection_closed_handler=self._connection_closed_handler, connection_name="rstream-locator"
+            connection_closed_handler=self._on_close_handler, connection_name="rstream-locator"
         )
         self.super_stream_metadata = DefaultSuperstreamMetadata(self.super_stream, self._default_client)
         if self.routing == RouteType.Hash:
@@ -161,3 +161,7 @@ async def reconnect_stream(self, stream: str) -> None:
         # close previous clients and re-create a publisher (with a new client)
         if self._producer is not None:
             await self._producer.reconnect_stream(stream)
+
+    async def stream_exists(self, stream: str) -> bool:
+        producer = await self._get_producer()
+        return await producer.stream_exists(stream)
diff --git a/rstream/utils.py b/rstream/utils.py
index 0944669..baca092 100644
--- a/rstream/utils.py
+++ b/rstream/utils.py
@@ -45,7 +45,7 @@ def __await__(self) -> Generator[Any, None, Any]:
 
 
 @dataclass
-class DisconnectionErrorInfo:
+class OnClosedErrorInfo:
     reason: str
     streams: list[str]
 


Test patch:
diff --git a/tests/test_consumer.py b/tests/test_consumer.py
index c87c054..f0a6c24 100644
--- a/tests/test_consumer.py
+++ b/tests/test_consumer.py
@@ -13,10 +13,10 @@
     AMQPMessage,
     Consumer,
     ConsumerOffsetSpecification,
-    DisconnectionErrorInfo,
     FilterConfiguration,
     MessageContext,
     OffsetType,
+    OnClosedErrorInfo,
     Producer,
     RouteType,
     SuperStreamConsumer,
@@ -33,6 +33,7 @@
     routing_extractor_generic,
     run_consumer,
     task_to_delete_connection,
+    task_to_delete_stream_consumer,
     wait_for,
 )
 
@@ -538,7 +539,7 @@ async def test_consumer_connection_broke(stream: str) -> None:
     stream_disconnected = None
     consumer_broke: Consumer
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         nonlocal connection_broke
         connection_broke = True
         nonlocal consumer_broke
@@ -553,7 +554,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         vhost="/",
         username="guest",
         password="guest",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         connection_name="test-connection",
     )
 
@@ -578,7 +579,7 @@ async def test_super_stream_consumer_connection_broke(super_stream: str) -> None
     streams_disconnected: set[str] = set()
     consumer_broke: Consumer
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         nonlocal connection_broke
         nonlocal streams_disconnected
         # avoiding multiple connection closed to hit
@@ -600,7 +601,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         vhost="/",
         username="guest",
         password="guest",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         connection_name="test-connection",
         super_stream=super_stream,
     )
@@ -629,7 +630,7 @@ async def test_super_stream_consumer_connection_broke_with_reconnect(super_strea
     consumer_broke: Consumer
     offset_restart = 0
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         logger.warning("connection closed")
         nonlocal connection_broke
         nonlocal streams_disconnected
@@ -654,7 +655,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         password="guest",
         routing_extractor=routing_extractor_generic,
         routing=RouteType.Hash,
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         connection_name="test-connection",
         super_stream=super_stream,
     )
@@ -677,7 +678,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         vhost="/",
         username="guest",
         password="guest",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         connection_name="test-connection",
         super_stream=super_stream,
     )
@@ -780,3 +781,46 @@ async def on_message(msg: AMQPMessage, message_context: MessageContext):
 
     # No filter on produce side no filetering
     await wait_for(lambda: len(captured) == 0)
+
+
+async def test_consumer_metadata_update(consumer: Consumer) -> None:
+
+    consumer_closed = False
+    stream_disconnected = None
+    stream = "test-stream-metadata-update"
+    consumer_metadata_update: Consumer
+
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
+        nonlocal consumer_closed
+
+        nonlocal consumer_metadata_update
+        nonlocal stream_disconnected
+        stream_disconnected = disconnection_info.streams.pop()
+
+        if consumer_closed is False:
+            consumer_closed = True
+            await consumer_metadata_update.close()
+
+    consumer_metadata_update = Consumer(
+        host="localhost",
+        port=5552,
+        vhost="/",
+        username="guest",
+        password="guest",
+        on_close_handler=on_connection_closed,
+        connection_name="test-connection",
+    )
+
+    async def on_message(msg: AMQPMessage, message_context: MessageContext):
+        message_context.consumer.get_stream(message_context.subscriber_name)
+
+    await consumer_metadata_update.start()
+    await consumer_metadata_update.create_stream(stream)
+    asyncio.create_task(task_to_delete_stream_consumer(consumer, stream))
+    await consumer_metadata_update.subscribe(stream=stream, callback=on_message, decoder=amqp_decoder)
+    await consumer_metadata_update.run()
+
+    assert consumer_closed is True
+    assert stream_disconnected == stream
+
+    await asyncio.sleep(1)
diff --git a/tests/test_producer.py b/tests/test_producer.py
index a3676e7..c2fe808 100644
--- a/tests/test_producer.py
+++ b/tests/test_producer.py
@@ -11,7 +11,7 @@
     AMQPMessage,
     CompressionType,
     Consumer,
-    DisconnectionErrorInfo,
+    OnClosedErrorInfo,
     Producer,
     RawMessage,
     RouteType,
@@ -26,6 +26,7 @@
     on_publish_confirm_client_callback2,
     routing_extractor_generic,
     task_to_delete_connection,
+    task_to_delete_stream_producer,
     wait_for,
 )
 
@@ -478,7 +479,7 @@ async def test_producer_connection_broke(stream: str) -> None:
     stream_disconnected = None
     producer_broke: Producer
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         nonlocal connection_broke
         connection_broke = True
         nonlocal producer_broke
@@ -492,7 +493,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         "localhost",
         username="guest",
         password="guest",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         connection_name="test-connection",
     )
 
@@ -518,7 +519,7 @@ async def test_producer_connection_broke_with_send_batch(stream: str) -> None:
     stream_disconnected = None
     producer_broke: Producer
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
 
         nonlocal connection_broke
         connection_broke = True
@@ -531,7 +532,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         "localhost",
         username="guest",
         password="guest",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         connection_name="test-connection",
     )
 
@@ -566,7 +567,7 @@ async def test_super_stream_producer_connection_broke(super_stream: str) -> None
     streams_disconnected: set[str] = set()
     producer_broke: Producer
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         nonlocal connection_broke
         connection_broke = True
         nonlocal producer_broke
@@ -581,7 +582,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         password="guest",
         routing_extractor=routing_extractor_generic,
         routing=RouteType.Hash,
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         connection_name="test-connection",
         super_stream=super_stream,
     )
@@ -621,7 +622,7 @@ async def test_producer_connection_broke_with_reconnect(stream: str) -> None:
     stream_disconnected = None
     producer_broke: Producer
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         nonlocal connection_broke
         connection_broke = True
         nonlocal producer_broke
@@ -637,7 +638,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         "localhost",
         username="guest",
         password="guest",
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         connection_name="test-connection",
     )
 
@@ -671,7 +672,7 @@ async def test_super_stream_producer_connection_broke_with_reconnect(super_strea
     streams_disconnected: set[str] = set()
     producer_broke: Producer
 
-    async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> None:
+    async def on_connection_closed(disconnection_info: OnClosedErrorInfo) -> None:
         nonlocal connection_broke
         connection_broke = True
         nonlocal producer_broke
@@ -687,7 +688,7 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
         password="guest",
         routing_extractor=routing_extractor_generic,
         routing=RouteType.Hash,
-        connection_closed_handler=on_connection_closed,
+        on_close_handler=on_connection_closed,
         connection_name="test-connection",
         super_stream=super_stream,
     )
@@ -722,3 +723,42 @@ async def on_connection_closed(disconnection_info: DisconnectionErrorInfo) -> No
     assert "test-super-stream-0" in streams_disconnected
     assert "test-super-stream-1" in streams_disconnected
     assert "test-super-stream-2" in streams_disconnected
+
+
+async def test_producer_metadata_update(producer: Producer) -> None:
+
+    producer_closed = False
+    stream_updated = None
+    producer_metadata: Producer
+    stream_metadata_updated = "test-metadata-update-stream"
+
+    async def on_metadata_update(on_closed_info: OnClosedErrorInfo) -> None:
+
+        nonlocal producer_closed
+        producer_closed = True
+        nonlocal stream_updated
+        stream_updated = on_closed_info.streams[0]
+        nonlocal producer_metadata
+        await producer_metadata.close()
+
+    async with Producer(
+        "localhost",
+        username="guest",
+        password="guest",
+        on_close_handler=on_metadata_update,
+    ) as producer_metadata:
+
+        await producer_metadata.create_stream(stream_metadata_updated, exists_ok=True)
+        # delete the stream after a while
+        asyncio.create_task(task_to_delete_stream_producer(producer, stream_metadata_updated))
+
+        while producer_closed is False:
+            try:
+                await producer_metadata.send(stream_metadata_updated, b"one")
+            # Connection broke
+            except BaseException:
+                producer_closed = True
+                break
+
+    assert producer_closed is True
+    assert stream_updated == stream_metadata_updated
diff --git a/tests/util.py b/tests/util.py
index 538d98d..8f19356 100644
--- a/tests/util.py
+++ b/tests/util.py
@@ -9,10 +9,12 @@
 from rstream import (
     AMQPMessage,
     ConfirmationStatus,
+    Consumer,
     EventContext,
     MessageContext,
     OffsetSpecification,
     OffsetType,
+    Producer,
     SuperStreamConsumer,
     amqp_decoder,
 )
@@ -139,5 +141,21 @@ async def task_to_delete_connection(connection_name: str) -> None:
             await wait_for(lambda: get_connection(connection["name"]) is False)
 
 
+async def task_to_delete_stream_producer(producer: Producer, stream: str) -> None:
+
+    # delay a few seconds before deleting the connection
+    await asyncio.sleep(4)
+
+    await producer.delete_stream(stream)
+
+
+async def task_to_delete_stream_consumer(consumer: Consumer, stream: str) -> None:
+
+    # delay a few seconds before deleting the connection
+    await asyncio.sleep(4)
+
+    await consumer.delete_stream(stream)
+
+
 async def filter_value_extractor(message: AMQPMessage) -> str:
     return message.application_properties["id"]