Kafka & Event Streaming
Relevant source files
The following files were used as context for generating this wiki page:
The Kafka and Event Streaming system in Aether-X Ultimate serves as the backbone for real-time data ingestion, asynchronous processing, and service decoupling. It enables the platform to handle high-throughput data streams for ETL pipelines, monitoring, and content generation workflows. The system is integrated primarily through the data-processing service, which acts as the consumer and orchestrator for streaming jobs.
Sources: README.md:16-16, services/data-processing/app.py:1-10
Architecture Overview
The event streaming architecture is built on the Confluent Platform stack, utilizing Apache Kafka for message storage and Zookeeper for cluster coordination. The infrastructure includes specialized components for schema management and external connectivity.
Core Infrastructure Components
| Component | Image / Technology | Purpose |
|---|---|---|
| Zookeeper | confluentinc/cp-zookeeper:7.4.0 | Cluster coordination and metadata management. |
| Kafka Broker | confluentinc/cp-kafka:7.4.0 | Central message broker for event persistence. |
| Schema Registry | confluentinc/cp-schema-registry:7.4.0 | Manages Avro/JSON schemas for message validation. |
| Kafka Connect | confluentinc/cp-kafka-connect:7.4.0 | Framework for connecting Kafka to external databases/APIs. |
| Kafka REST Proxy | confluentinc/cp-kafka-rest:7.4.0 | Provides a RESTful interface to the Kafka cluster. |
| Kafka UI | provectuslabs/kafka-ui:latest | Web interface for monitoring topics and consumers. |
Sources: infrastructure/databases/docker-compose.yml:104-142, infrastructure/kafka/docker-compose.yml:6-100
Streaming Data Flow
The following diagram illustrates how data flows from external clients through the API Gateway into the Kafka-backed data processing pipeline.
The system uses a combination of in-memory queuing for immediate ingestion and dedicated background threads for stream processing logic.
Sources: services/data-processing/app.py:41-65, services/api-gateway/app.py:168-176, infrastructure/kafka/docker-compose.yml:41-62
Data Processing Service Integration
The data-processing service (running on port 8006) provides the primary API for interacting with the event stream. It supports data ingestion, ETL (Extract, Transform, Load) pipelines, and long-running stream processors.
Ingestion API
Data is ingested via a POST request, which assigns a unique message ID and timestamps the payload before entering the processing queue.
- Endpoint:
POST /api/v1/data/ingest - Request Body:
{ "type": "string", "data": "object" }
Sources: services/data-processing/app.py:19-40, services/api-gateway/app.py:168-176
Stream Processors
The service manages dynamic "Stream Processors" which are background threads that continuously consume and transform data based on a provided configuration.
# services/data-processing/app.py:157-189
def process_stream(processor_id, config):
"""Process data stream in background"""
while True:
if stream_processors.get(processor_id, {}).get('status') == 'stopped':
break
if not data_queue.empty():
message = data_queue.get(timeout=1)
processed_message = {
'original_id': message['id'],
'processor_id': processor_id,
'processed_data': apply_stream_processing(message['data'], config),
'processed_at': datetime.utcnow().isoformat()
}
processed_data.append(processed_message)
Sources: services/data-processing/app.py:157-189
Configuration and Connectivity
In development environments, Kafka is configured to listen on localhost:9092 with Zookeeper on 2181. The KAFKA_ADVERTISED_LISTENERS environment variable is critical for ensuring microservices can reach the broker within the Docker network.
| Variable | Value | Description |
|---|---|---|
KAFKA_BROKER_ID | 1 | Unique identifier for the broker. |
KAFKA_ZOOKEEPER_CONNECT | zookeeper:2181 | Connection string for Zookeeper. |
SCHEMA_REGISTRY_URL | http://schema-registry:8081 | URL used by REST Proxy and UI for schema lookup. |
CONNECT_BOOTSTRAP_SERVERS | kafka:9092 | Broker connection for Kafka Connect. |
Sources: infrastructure/databases/docker-compose.yml:128-135, infrastructure/kafka/docker-compose.yml:45-48, config/development.yaml:49-51
ETL Pipeline Execution
The system supports synchronous ETL pipelines where data is extracted from sources (API or Database), transformed (filtering or aggregation), and loaded into a destination (Memory or external storage).
The logic for these steps is defined in the execute_etl_pipeline function, which iterates through a list of step definitions.
Sources: services/data-processing/app.py:133-155, services/data-processing/app.py:191-235
Summary
Kafka and Event Streaming in Aether-X Ultimate provide a scalable foundation for handling real-time data. By leveraging the Confluent stack alongside a dedicated Python-based processing service, the platform ensures that data-intensive AI operations and ETL tasks are handled asynchronously and reliably. This setup allows for complex data enrichment and transformation before results are persisted to the various SQL and NoSQL databases in the cluster.