The is an module designed for seamless integration with Apache Kafka, a distributed event streaming platform. This module allows to function as a consumer and producer of messages on a Kafka broker, facilitating real-time data exchange with other business applications and services. It is built for enterprise environments that require high performance and reliability for mission-critical applications.
Real-time Message Processing: The module continuously consumes messages from configured Kafka topics, ensuring that business processes are updated in real-time.
Concurrent Processing: It supports the simultaneous execution of multiple operations in isolated processes, which improves resource utilization and allows for greater scalability.
Robust Error Handling: The connector includes configurable retry logic with exponential backoff and comprehensive error recovery mechanisms to ensure message delivery and processing.
Health Monitoring: It provides periodic health checks and detailed status reporting, which can be monitored via logging system.
Comprehensive Message Logging: The module keeps a detailed audit trail and a complete history of all processed messages, which is essential for tracking and troubleshooting.
Smart Notifications: An intelligent notification system alerts administrators to potential issues with detailed error reports.
Bidirectional Communication: The module supports both consuming messages from Kafka and producing messages to Kafka, providing a true two-way integration.
Scalable Architecture: Designed with enterprise-grade reliability and scalability, it is suitable for high-volume data processing and mission-critical applications.
Kafka Connector
Enterprise-Grade Kafka Integration
🚀 Real-time Message Processing
Seamlessly integrate Kafka message streams with your business processes. Built for enterprise scalability and reliability.
✨ Key Features
Real-time Consumption
Continuously consume messages from configured Kafka topics with high-performance streaming capabilities.
Concurrent Processing
Executes multiple operations simultaneously in isolated processes, ensuring efficient resource usage and scalability.
Error Handling & Retry
Configurable retry logic with exponential backoff and comprehensive error recovery mechanisms.
Health Monitoring
Periodic health checks with detailed status reporting and performance metrics tracking, output to the server console via logging system.
Message Logging
Comprehensive message tracking and audit trail with detailed processing history.
Smart Notifications
Intelligent notification system for administrators with detailed error reporting and alerts.
🏗️ Architecture Overview
🚀 Quick Start
1. Install Dependencies
2. Set DB Name (.conf)
3. Configure Master Consumer
Navigate to Kafka Connector → Master Consumers
4. Monitoring Message Logs
Navigate to Kafka Connector → Message Logs
5. Set Target Model & Function
def consume_purchase_order(self, message): # Dummy method for simulation data vals = { 'origin': message.get('order_ref'), 'partner_id': message.get('vendor_name'), 'order_line': [ (0, 0, { 'product_id': item.get('product'), 'product_qty': item.get('qty'), 'price_unit': item.get('price'), }) for item in message.get('items', []) ] } try: create_po = self.with_user(2).create(vals) return create_po except Exception as e: raise ValidationError(_(f'Failed to consume purchase order. Error: {e}'))
6. Start Consumer
To start the Kafka consumer, click the RUN button located in the Master Consumer form or from the Tree View panel.
7. Stop Consumer
To stop the Kafka consumer, click the STOP button located in the Master Consumer form or from the Tree View panel.
8. Produce Message
from import models, fields, api from kafka import KafkaProducer from json import dumps class PoInherit(models.Model): _inherit = 'purchase.order' @api.model def create(self, values): producer = KafkaProducer(bootstrap_servers=['localhost:9092'], value_serializer=lambda x: dumps(x).encode('utf-8')) for rec in self: data = { 'partner_id': rec.id } producer.send('po_kafka', value=data) producer.flush() return super(PoInherit, self).create(values)
🛠️ Technology Stack
🎯 Use Cases
Order Processing
Real-time order updates from e-commerce platforms and external systems.
Analytics & Reporting
Stream analytics data and generate real-time business intelligence reports.
System Integration
Seamless integration between multiple business systems and applications.
IoT Data Processing
Process sensor data and IoT device messages in real-time for smart operations.
📸 Screenshots
Master Consumer Management
Configure and manage your Kafka consumer settings with an intuitive interface.
Message Processing Logs
Monitor all processed messages with detailed logs and status information.
Real-time Control Panel
Start, stop, and monitor your Kafka consumers with real-time status updates.
Performance Metrics
Track performance metrics, and error rates in real-time.
🏆 Enterprise Ready
Built with enterprise-grade reliability, scalability, and security in mind. Perfect for high-volume message processing and mission-critical applications.