Enabling 20MB message for python sdk clients for Eventhubs

sdk/eventhub/azure-eventhub/azure/eventhub/_producer_client.py

@@ -359,6 +359,8 @@ def _get_max_message_size(self) -> None:
                     )
                     or self._amqp_transport.MAX_MESSAGE_LENGTH_BYTES
                 )
+                print(f">>> _get_max_message_size: _max_message_size_on_link = {self._max_message_size_on_link}")
+                print(f">>> _get_max_message_size: MAX_MESSAGE_LENGTH_BYTES = {self._amqp_transport.MAX_MESSAGE_LENGTH_BYTES}")
 
     def _start_producer(self, partition_id: str, send_timeout: Optional[float] = None) -> None:
         with self._lock:
@@ -797,6 +799,9 @@ def create_batch(  # pylint: disable=unused-argument
         """
         if not self._max_message_size_on_link:
             self._get_max_message_size()
+
+        print(f">>> create_batch: max_size_in_bytes parameter = {max_size_in_bytes}")
+        print(f">>> create_batch: _max_message_size_on_link = {self._max_message_size_on_link}")
 
         if max_size_in_bytes and max_size_in_bytes > self._max_message_size_on_link:
             raise ValueError(

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/client.py

@@ -576,7 +576,7 @@ class SendClient(AMQPClient):
     def __init__(self, hostname, target, **kwargs):
         self.target = target
         # Sender and Link settings
-        self._max_message_size = kwargs.pop("max_message_size", MAX_FRAME_SIZE_BYTES)
+        self._max_message_size = kwargs.pop("max_message_size", 0)
         self._link_properties = kwargs.pop("link_properties", None)
         self._link_credit = kwargs.pop("link_credit", None)
         super(SendClient, self).__init__(hostname, **kwargs)

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/link.py

@@ -7,7 +7,7 @@
 from typing import Any, Optional, TYPE_CHECKING
 import uuid
 import logging
-
+logging.basicConfig(level=logging.INFO)
 from .error import AMQPError, ErrorCondition, AMQPLinkError, AMQPLinkRedirect, AMQPConnectionError
 from .endpoints import Source, Target
 from .constants import DEFAULT_LINK_CREDIT, SessionState, LinkState, Role, SenderSettleMode, ReceiverSettleMode
@@ -161,6 +161,7 @@ def _outgoing_attach(self) -> None:
             desired_capabilities=self.desired_capabilities if self.state == LinkState.DETACHED else None,
             properties=self.properties,
         )
+        _LOGGER.info(f"Sending Attach frame with max_message_size={self.max_message_size}")
         if self.network_trace:
             _LOGGER.debug("-> %r", attach_frame, extra=self.network_trace_params)
         self._session._outgoing_attach(attach_frame)  # pylint: disable=protected-access

sdk/eventhub/azure-eventhub/samples/sync_samples/send.py

@@ -94,6 +94,22 @@ def send_event_data_list(producer):
     except EventHubError as eh_err:
         print("Sending error: ", eh_err)
 
+def send_single_large_message(producer, size_mb=17):
+    """
+    Try to send one single oversized message to Event Hub.
+    Now with explicit large max_size_in_bytes to allow 20MB messages.
+    """
+    payload = "X" * (size_mb * 1024 * 1024)  # 20 MB string
+    try:
+        # Explicitly set max_size_in_bytes to allow large messages
+        batch = producer.create_batch(max_size_in_bytes=19 * 1024 * 1024)  # 20MB limit
+        batch.add(EventData(payload))
+        producer.send_batch(batch)
+        print(f"SUCCESS: sent {size_mb} MB message!")
+    except ValueError as e:
+        print(f"Failed to add {size_mb} MB message to batch -> {e}")
+    except Exception as e:
+        print(f"Failed to send {size_mb} MB message -> {e}")
 
 def send_concurrent_with_shared_client_and_lock():
     """
@@ -139,6 +155,7 @@ def send_with_lock(thread_id):
     send_event_data_batch_with_partition_id(producer)
     send_event_data_batch_with_properties(producer)
     send_event_data_list(producer)
+    send_single_large_message(producer, size_mb=17)
 
 print("Send messages in {} seconds.".format(time.time() - start_time))
 

sdk/eventhub/azure-eventhub/tests/conftest.py

@@ -11,10 +11,10 @@
     get_region_override,
     get_credential as get_devtools_credential,
 )
-from azure.eventhub.extensions.checkpointstoreblobaio import (
-    BlobCheckpointStore as BlobCheckpointStoreAsync,
-)
-from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
+# from azure.eventhub.extensions.checkpointstoreblobaio import (
+#     BlobCheckpointStore as BlobCheckpointStoreAsync,
+# )
+# from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
 from azure.eventhub._pyamqp.authentication import SASTokenAuth
 from azure.eventhub._pyamqp import ReceiveClient
 from azure.eventhub import EventHubProducerClient, TransportType