Update AsyncIO producer architecture to improve performance

.gitignore

@@ -34,3 +34,5 @@ venv_examples
 .coverage
 **/coverage.xml
 **/test-report.xml
+.ducktape
+results

aio_producer_simple_diagram.md

@@ -0,0 +1,47 @@
+# AIOProducer Simple Class Diagram
+
+```
+┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
+│   AIOProducer   │    │BufferTimeout    │    │ MessageBatch    │
+│  (Orchestrator) │    │   Manager       │    │ (Value Object)  │
+├─────────────────┤    ├─────────────────┤    ├─────────────────┤
+│ • produce()     │───▶│ • timeout       │    │ • immutable     │
+│ • flush()       │    │   monitoring    │    │ • type safe     │
+│ • orchestrate   │    │ • mark_activity │    │ • clean data    │
+└─────────────────┘    └─────────────────┘    └─────────────────┘
+         │                       │                       ▲
+         ▼                       ▼                       │
+┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
+│ProducerBatch    │    │ProducerBatch    │    │CallbackManager  │
+│   Manager       │───▶│   Executor      │    │ (Unified Mgmt)  │
+├─────────────────┤    ├─────────────────┤    ├─────────────────┤
+│ • create_batches│    │ • execute_batch │    │ • sync/async    │
+│ • group topics  │    │ • thread pool   │    │ • object pool   │
+│ • manage buffer │    │ • poll results  │    │ • event loop    │
+└─────────────────┘    └─────────────────┘    └─────────────────┘
+         │                                             │
+         ▼                                             ▼
+┌─────────────────┐                          ┌─────────────────┐
+│create_message   │                          │ ReusableMessage │
+│   _batch()      │                          │ Callback Pool   │
+├─────────────────┤                          ├─────────────────┤
+│ • factory func  │                          │ • pooled objs   │
+│ • type safety   │◄─────────────────────────│ • auto-return   │
+│ • validation    │                          │ • thread safe   │
+└─────────────────┘                          └─────────────────┘
+```
+
+## Architecture Summary
+
+**7 Components Total:**
+- **1 Orchestrator**: AIOProducer (main API)
+- **3 Core Services**: ProducerBatchManager, ProducerBatchExecutor, BufferTimeoutManager  
+- **1 Unified Manager**: CallbackManager (merged handler + pool)
+- **2 Data Objects**: MessageBatch + factory function
+
+**Key Benefits:**
+- ✅ Single responsibility per component
+- ✅ Clean dependency injection  
+- ✅ Unified callback management
+- ✅ Immutable data structures
+- ✅ Performance optimized pooling

src/confluent_kafka/aio/_AIOConsumer.py

@@ -41,14 +41,7 @@ def __init__(self, consumer_conf, max_workers=2, executor=None):
         self._consumer = confluent_kafka.Consumer(consumer_conf)
 
     async def _call(self, blocking_task, *args, **kwargs):
-        return (await asyncio.gather(
-            asyncio.get_running_loop().run_in_executor(self.executor,
-                                                       functools.partial(
-                                                           blocking_task,
-                                                           *args,
-                                                           **kwargs))
-
-        ))[0]
+        return await _common.async_call(self.executor, blocking_task, *args, **kwargs)
 
     def _wrap_callback(self, loop, callback, edit_args=None, edit_kwargs=None):
         def ret(*args, **kwargs):

src/confluent_kafka/aio/__init__.py

@@ -13,5 +13,5 @@
 # limitations under the License.
 
 from ._AIOConsumer import AIOConsumer
-from ._AIOProducer import AIOProducer
+from .producer import AIOProducer
 __all__ = ['AIOConsumer', 'AIOProducer']

src/confluent_kafka/aio/_common.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import asyncio
+import functools
 
 
 class AsyncLogger:
@@ -48,6 +49,26 @@ def wrap_conf_logger(loop, conf):
         conf['logger'] = AsyncLogger(loop, conf['logger'])
 
 
+async def async_call(executor, blocking_task, *args, **kwargs):
+    """Helper function for blocking operations that need ThreadPool execution
+
+    Args:
+        executor: ThreadPoolExecutor to use for blocking operations
+        blocking_task: The blocking function to execute
+        *args, **kwargs: Arguments to pass to the blocking function
+
+    Returns:
+        Result of the blocking function execution
+    """
+    return (await asyncio.gather(
+        asyncio.get_running_loop().run_in_executor(executor,
+                                                   functools.partial(
+                                                       blocking_task,
+                                                       *args,
+                                                       **kwargs))
+    ))[0]
+
+
 def wrap_common_callbacks(loop, conf):
     wrap_conf_callback(loop, conf, 'error_cb')
     wrap_conf_callback(loop, conf, 'throttle_cb')