“

Series: S8 Technology Deep Dives · Article 5 | Level: Advanced | Reading Time: 20 min

Kafka's 7 Usage Patterns: From Event Sourcing to Stream-Batch Unification

#TL;DR

Apache Kafka serves 7 core roles in coomia-dip: event bus, CDC transport, stream processing pipeline, cross-Layer communication, audit logging, metrics collection, and command queue
Through carefully designed topic naming conventions, partitioning strategies, and consumer group management, coomia-dip achieves millisecond event propagation, exactly-once semantics, and multi-tenant isolation
This article details the architecture design for each pattern, configuration best practices, performance tuning parameters, and integration with Flink, Iceberg, and Doris

#1. Kafka's Panoramic Role in coomia-dip

#1.1 Why Kafka

In coomia-dip's 8-Layer architecture, data flows everywhere: Ontology changes need real-time propagation, CDC data needs reliable transport, cross-Layer async communication needs decoupling, and audit events need durable storage. Kafka's unique log-based model perfectly matches these requirements:

Requirement	Kafka Feature	Alternative (and Limitation)
High throughput	Sequential disk writes, zero-copy	RabbitMQ (10x lower throughput)
Durability	Messages persisted to disk	Redis Streams (memory-bound)
Replay	Consumers can re-consume from any offset	RabbitMQ (consumed = deleted)
Multiple consumers	Consumer groups independently consume same topic	Pulsar (higher operational complexity)
Exactly-once	Transactions + idempotent producers	Most MQs only support at-least-once
Stream integration	Native Flink/Spark Structured Streaming	Requires additional adapter layer

#1.2 Kafka Deployment Architecture

Code

┌────────────────────────────────────────────────────────┐
│                  coomia-dip Kafka Cluster                │
│                                                         │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐                │
│  │Broker 1 │  │Broker 2 │  │Broker 3 │                │
│  │(KRaft)  │  │(KRaft)  │  │(KRaft)  │                │
│  └────┬────┘  └────┬────┘  └────┬────┘                │
│       └────────────┼────────────┘                      │
│                    │                                    │
│  ┌─────────────────▼────────────────────────────────┐  │
│  │              Topic Layout                         │  │
│  │                                                   │  │
│  │  ontology.events.*     ← Event Bus (Pattern 1)    │  │
│  │  cdc.{source}.*        ← CDC Transport (Pattern 2)│  │
│  │  stream.{pipeline}.*   ← Streaming (Pattern 3)    │  │
│  │  Layer.{src}.{dst}.*   ← Cross-Layer (Pattern 4)  │  │
│  │  audit.{Layer}.*       ← Audit Log (Pattern 5)    │  │
│  │  metrics.{Layer}.*     ← Metrics (Pattern 6)      │  │
│  │  command.{service}.*   ← Command Queue (Pattern 7) │  │
│  └──────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────┘

#1.3 Topic Naming Convention

Code

{domain}.{category}.{entity}.{version}

Examples:
  ontology.events.object-type.v1      # Ontology object type change events
  cdc.mysql.orders.v1                  # MySQL orders table CDC stream
  stream.pipeline.etl-bronze.v1        # ETL bronze layer data stream
  Layer.b.d.reasoning-request.v1       # Control → Reasoning requests
  audit.Layer-b.access-log.v1          # Control Layer access audit log
  metrics.Layer-c.query-latency.v1     # Data Layer query latency metrics
  command.action-engine.execute.v1     # Action engine execution commands

#2. Pattern 1: Ontology Event Bus

#2.1 Design Goals

When object types, relation types, or instances change in the Ontology, all downstream systems must be notified in real time. The event bus pattern models Ontology changes as immutable events, broadcasting them through Kafka to all subscribers.

#2.2 Event Schema Design

Python

# intelligence-Layer/ontology_events/schemas.py
from pydantic import BaseModel
from datetime import datetime
from enum import Enum
from typing import Any

class EventType(str, Enum):
    OBJECT_TYPE_CREATED = "object_type.created"
    OBJECT_TYPE_UPDATED = "object_type.updated"
    OBJECT_TYPE_DELETED = "object_type.deleted"
    OBJECT_INSTANCE_CREATED = "object_instance.created"
    OBJECT_INSTANCE_UPDATED = "object_instance.updated"
    OBJECT_INSTANCE_DELETED = "object_instance.deleted"
    RELATION_CREATED = "relation.created"
    RELATION_DELETED = "relation.deleted"

class OntologyEvent(BaseModel):
    """Ontology change event"""
    event_id: str
    event_type: EventType
    tenant_id: str
    world_id: str
    entity_rid: str
    entity_type: str
    timestamp: datetime
    payload: dict[str, Any]
    metadata: dict[str, str]
    causation_id: str | None = None
    correlation_id: str | None = None

#2.3 Producer Implementation

Python

# control-Layer/ontology/event_publisher.py
from confluent_kafka import Producer

class OntologyEventPublisher:
    """Ontology event publisher"""

    def __init__(self, bootstrap_servers: str):
        self.producer = Producer({
            "bootstrap.servers": bootstrap_servers,
            "acks": "all",                    # All replicas acknowledge
            "enable.idempotence": True,       # Idempotent producer
            "max.in.flight.requests.per.connection": 5,
            "retries": 10,
            "retry.backoff.ms": 100,
            "compression.type": "zstd",       # High compression ratio
            "linger.ms": 5,                   # Batch sending
            "batch.size": 65536,              # 64KB batch
        })

    def publish(self, event: OntologyEvent):
        """Publish an Ontology event"""
        topic = f"ontology.events.{event.entity_type}.v1"

        self.producer.produce(
            topic=topic,
            key=event.entity_rid.encode("utf-8"),
            value=event.model_dump_json().encode("utf-8"),
            headers={
                "event_type": event.event_type.value,
                "tenant_id": event.tenant_id,
                "correlation_id": event.correlation_id or "",
            },
            callback=self._delivery_callback,
        )
        self.producer.flush()

    def _delivery_callback(self, err, msg):
        if err:
            logger.error(f"Event delivery failed: {err}")
        else:
            logger.debug(
                f"Event delivered to {msg.topic()}[{msg.partition()}]@{msg.offset()}"
            )

#2.4 Consumer Implementation

Python

# data-Layer/ontology/event_consumer.py
from confluent_kafka import Consumer

class OntologyEventConsumer:
    """Ontology event consumer"""

    def __init__(self, bootstrap_servers: str, group_id: str):
        self.consumer = Consumer({
            "bootstrap.servers": bootstrap_servers,
            "group.id": group_id,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,           # Manual commit
            "max.poll.interval.ms": 300000,        # 5-min processing timeout
            "session.timeout.ms": 45000,
            "heartbeat.interval.ms": 15000,
            "isolation.level": "read_committed",   # Transactional isolation
        })

    def consume_loop(self, topics: list[str], handler):
        """Main consumption loop"""
        self.consumer.subscribe(topics)

        try:
            while True:
                msg = self.consumer.poll(timeout=1.0)
                if msg is None:
                    continue
                if msg.error():
                    logger.error(f"Consumer error: {msg.error()}")
                    continue

                event = OntologyEvent.model_validate_json(msg.value())

                try:
                    handler(event)
                    self.consumer.commit(msg)
                except Exception as e:
                    logger.error(f"Event processing failed: {e}")
                    # Do not commit — message will be redelivered
        finally:
            self.consumer.close()

#3. Pattern 2: CDC Transport Channel

#3.1 CDC Flow Architecture

Code

┌──────────────┐      ┌──────────────┐      ┌──────────────┐
│  MySQL/PG    │      │    Kafka     │      │   Flink CDC  │
│  (Source DB) │─CDC─▶│  cdc.*.v1    │─────▶│   Processor  │
│              │      │              │      │              │
└──────────────┘      └──────────────┘      └──────┬───────┘
                                                    │
                                        ┌───────────┼───────────┐
                                        ▼           ▼           ▼
                                   ┌────────┐  ┌────────┐  ┌────────┐
                                   │Iceberg │  │ Doris  │  │ Search │
                                   │(Cold)  │  │(Hot)   │  │(Index) │
                                   └────────┘  └────────┘  └────────┘

#3.2 Debezium Connector Configuration

JSON

{
  "name": "mysql-cdc-orders",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-source",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "${MYSQL_CDC_PASSWORD}",
    "database.server.id": "1001",
    "topic.prefix": "cdc.mysql",
    "database.include.list": "business_db",
    "table.include.list": "business_db.orders,business_db.customers",
    "schema.history.internal.kafka.topic": "cdc.schema-history",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "cdc\\.mysql\\.business_db\\.(.*)",
    "transforms.route.replacement": "cdc.mysql.$1.v1",
    "snapshot.mode": "initial",
    "topic.creation.default.replication.factor": 3,
    "topic.creation.default.partitions": 6
  }
}

#3.3 CDC Topic Special Configuration

PROPERTIES

# CDC topics need longer retention (support full snapshot rebuilds)
retention.ms=604800000        # 7 days
cleanup.policy=compact,delete  # Log compaction + expiry deletion
min.compaction.lag.ms=3600000  # 1-hour compaction delay
segment.ms=3600000             # Hourly segment rollover

#4. Pattern 3: Stream Processing Pipeline

#4.1 Three-Layer Data Pipeline

Code

Raw Events (Bronze)  →  Cleaned (Silver)  →  Aggregated (Gold)
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│ stream.pipe  │     │ stream.pipe  │     │ stream.pipe  │
│ .bronze.v1   │────▶│ .silver.v1   │────▶│ .gold.v1     │
└──────────────┘     └──────────────┘     └──────────────┘
      │                    │                    │
   Flink Job 1         Flink Job 2         Flink Job 3
  (Clean + Dedup)    (Enrich + Normalize) (Aggregate + Materialize)

#4.2 Flink + Kafka Stream Processing

Java

// data-Layer/flink/KafkaStreamPipeline.java
public class KafkaStreamPipeline {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(30000);
        env.getCheckpointConfig().setCheckpointingMode(
            CheckpointingMode.EXACTLY_ONCE);

        // Bronze: read from raw topic
        KafkaSource<String> bronzeSource = KafkaSource.<String>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("stream.pipeline.bronze.v1")
            .setGroupId("flink-bronze-processor")
            .setStartingOffsets(OffsetsInitializer.committedOffsets(
                OffsetResetStrategy.EARLIEST))
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();

        DataStream<String> bronzeStream = env.fromSource(
            bronzeSource, WatermarkStrategy.noWatermarks(), "Bronze Source"
        );

        // Clean + deduplicate
        DataStream<CleanedEvent> silverStream = bronzeStream
            .map(new EventParser())
            .filter(new DataQualityFilter())
            .keyBy(event -> event.getEntityId())
            .process(new DeduplicationProcessor(Duration.ofMinutes(5)));

        // Write to Silver topic
        KafkaSink<CleanedEvent> silverSink = KafkaSink.<CleanedEvent>builder()
            .setBootstrapServers("kafka:9092")
            .setRecordSerializer(
                KafkaRecordSerializationSchema.builder()
                    .setTopic("stream.pipeline.silver.v1")
                    .setKeySerializationSchema(new EventKeySerializer())
                    .setValueSerializationSchema(new EventValueSerializer())
                    .build()
            )
            .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            .setTransactionalIdPrefix("flink-silver")
            .build();

        silverStream.sinkTo(silverSink);
        env.execute("Bronze to Silver Pipeline");
    }
}

#5. Pattern 4: Cross-Layer Async Communication

#5.1 Request/Response Pattern

Code

Control Layer (Control)                    Reasoning & Decision Layer (Reasoning)
┌────────────────┐                  ┌────────────────┐
│                │  request topic   │                │
│  Send Request ─┼──────────────▶──┼─ Process       │
│                │                  │                │
│  Receive Resp ◀┼──────────────◀──┼─ Send Response │
│                │  response topic  │                │
└────────────────┘                  └────────────────┘

Topics:
  Layer.b.d.reasoning-request.v1   (partitioned by tenant_id)
  Layer.d.b.reasoning-response.v1  (partitioned by request_id)

#5.2 Implementation

Python

# control-Layer/cross_plane/kafka_rpc.py
import asyncio
from confluent_kafka import Producer

class KafkaAsyncRPC:
    """Kafka-based cross-Layer async RPC"""

    def __init__(self, bootstrap_servers: str, source_plane: str):
        self.producer = Producer({
            "bootstrap.servers": bootstrap_servers,
            "acks": "all",
            "enable.idempotence": True,
        })
        self.source_plane = source_plane
        self.pending_requests: dict[str, asyncio.Future] = {}

    async def call(
        self, target_plane: str, operation: str,
        payload: dict, timeout: float = 30.0,
    ) -> dict:
        """Async RPC call"""
        request_id = str(uuid.uuid4())
        topic = f"Layer.{self.source_plane}.{target_plane}.{operation}.v1"

        future = asyncio.get_event_loop().create_future()
        self.pending_requests[request_id] = future

        self.producer.produce(
            topic=topic,
            key=request_id.encode(),
            value=json.dumps({
                "request_id": request_id,
                "operation": operation,
                "payload": payload,
                "reply_topic": (
                    f"Layer.{target_plane}.{self.source_plane}"
                    f".{operation}-response.v1"
                ),
            }).encode(),
        )
        self.producer.flush()

        try:
            return await asyncio.wait_for(future, timeout=timeout)
        except asyncio.TimeoutError:
            del self.pending_requests[request_id]
            raise TimeoutError(
                f"RPC call to {target_plane}.{operation} timed out"
            )

#6. Pattern 5: Audit Logging

#6.1 Audit Event Structure

Python

# control-Layer/audit/audit_event.py
class AuditEvent(BaseModel):
    """Audit event"""
    audit_id: str
    timestamp: datetime
    tenant_id: str
    user_id: str
    action: str          # CREATE, READ, UPDATE, DELETE
    resource_type: str   # object_type, relation_type, action_template
    resource_id: str
    Layer: str           # Layer-b, Layer-c, Layer-d
    ip_address: str
    user_agent: str
    request_details: dict
    response_status: int
    duration_ms: int

#6.2 Audit Topic Configuration

PROPERTIES

# Audit logs need the longest retention (compliance requirement)
retention.ms=31536000000       # 365 days
cleanup.policy=delete          # No compaction — preserve full history
segment.bytes=1073741824       # 1GB segments
min.insync.replicas=2          # At least 2 synchronized replicas
unclean.leader.election.enable=false  # Prevent unclean leader election

#6.3 Audit Log Sink to Iceberg

Python

# data-Layer/audit/audit_sink.py
class AuditLogSink:
    """Persist audit logs to Iceberg tables"""

    def __init__(self):
        self.catalog = load_catalog("nessie")
        self.consumer = Consumer({
            "bootstrap.servers": "kafka:9092",
            "group.id": "audit-iceberg-sink",
            "enable.auto.commit": False,
            "isolation.level": "read_committed",
        })

    def run(self):
        self.consumer.subscribe([
            "audit.Layer-b.access-log.v1",
            "audit.Layer-c.query-log.v1",
            "audit.Layer-d.reasoning-log.v1",
        ])
        buffer = []
        last_flush = time.time()

        while True:
            msg = self.consumer.poll(timeout=1.0)
            if msg and not msg.error():
                buffer.append(json.loads(msg.value()))

            # Flush every 10 seconds or every 1000 messages
            if len(buffer) >= 1000 or (
                time.time() - last_flush > 10 and buffer
            ):
                self._write_to_iceberg(buffer)
                self.consumer.commit()
                buffer.clear()
                last_flush = time.time()

    def _write_to_iceberg(self, events: list[dict]):
        table = self.catalog.load_table("audit_db.access_logs")
        df = pa.Table.from_pylist(events)
        table.append(df)

#7. Pattern 6: Metrics Collection and Pattern 7: Command Queue

#7.1 Metrics Collection

Python

# deployment-Layer/metrics/kafka_metrics.py
class KafkaMetricsCollector:
    """Collect platform metrics via Kafka"""

    def emit_metric(self, metric_name: str, value: float,
                    tags: dict[str, str]):
        topic = f"metrics.{tags.get('Layer', 'unknown')}.{metric_name}.v1"
        self.producer.produce(
            topic=topic,
            value=json.dumps({
                "metric": metric_name,
                "value": value,
                "timestamp": datetime.utcnow().isoformat(),
                "tags": tags,
            }).encode(),
        )

Metrics topics use short retention (24-48 hours) and high partition counts (12+) to handle bursty metric emissions from all Layers. A dedicated Flink job aggregates raw metrics into time-windowed summaries before writing to Doris for dashboarding.

#7.2 Command Queue (CQRS Pattern)

Python

# control-Layer/command/command_queue.py
class CommandQueue:
    """Command queue — asynchronous write operations"""

    def submit_command(self, command_type: str, payload: dict) -> str:
        """Submit a command for async execution"""
        command_id = str(uuid.uuid4())
        topic = f"command.{command_type}.v1"

        self.producer.produce(
            topic=topic,
            key=command_id.encode(),
            value=json.dumps({
                "command_id": command_id,
                "type": command_type,
                "payload": payload,
                "submitted_at": datetime.utcnow().isoformat(),
                "status": "PENDING",
            }).encode(),
        )
        self.producer.flush()
        return command_id

    def get_command_status(self, command_id: str) -> str:
        """Query command execution status from the state store"""
        return self.state_store.get(f"command:{command_id}:status")

The CQRS pattern separates reads and writes. Write commands flow through Kafka to be processed asynchronously by dedicated command handlers. This decouples the API layer from heavy write operations (e.g., bulk Ontology imports, cross-table cascade updates), improving API responsiveness.

#8. Cluster Tuning and Operations

#8.1 Critical Broker Parameters

Parameter	Default	Recommended	Notes
`num.partitions`	1	6	Default partition count
`default.replication.factor`	1	3	Default replication factor
`min.insync.replicas`	1	2	Minimum in-sync replicas
`log.retention.hours`	168	Per-topic	Message retention
`log.segment.bytes`	1GB	512MB	Segment size
`compression.type`	producer	zstd	Compression algorithm
`message.max.bytes`	1MB	10MB	Max message size
`num.io.threads`	8	16	I/O thread count
`num.network.threads`	3	8	Network thread count

#8.2 Monitoring Metrics

Metric	Alert Threshold	Description
`UnderReplicatedPartitions`	> 0	Under-replicated partitions
`IsrShrinkRate`	> 0 (sustained)	ISR shrink rate
`RequestQueueSize`	> 100	Request queue depth
`ConsumerLag`	> 10000	Consumer lag
`ProduceRequestsPerSec`	Capacity-dependent	Production rate
`BytesInPerSec`	> 80% network bandwidth	Inbound byte rate

#8.3 Multi-Tenant Isolation

PROPERTIES

# Resource isolation via quotas
quota.producer.default=10485760  # 10MB/s per tenant
quota.consumer.default=20971520  # 20MB/s per tenant

# Higher quotas for specific tenants
# kafka-configs --alter --add-config 'producer_byte_rate=52428800'
#   --entity-type users --entity-name tenant-large

#9. Common Pitfalls and Solutions

#9.1 Consumer Rebalance Storms

Problem: Consumers frequently joining/leaving causes continuous rebalancing.

Solution:

Increase session.timeout.ms to 45s
Increase max.poll.interval.ms to 300s
Use Cooperative Sticky assignment strategy

#9.2 Message Backlog

Problem: Consumption speed cannot keep up with production speed.

Solution:

Increase partition count for higher parallelism
Optimize consumer processing logic
Consider replacing single-threaded consumers with Flink

#9.3 Message Loss

Problem: Broker failure causes message loss.

Solution:

Producer: set acks=all
Broker: set min.insync.replicas=2
Disable unclean leader election: unclean.leader.election.enable=false

#9.4 Message Ordering Issues

Problem: Events for the same entity arrive out of order.

Solution:

Use entity ID as message key to ensure same-entity messages land in the same partition
Producer: set max.in.flight.requests.per.connection=5 (with idempotent producer)

#Key Takeaways

One Kafka cluster, seven usage patterns: Through careful topic naming and differentiated configuration, coomia-dip uses a single Kafka cluster for event bus, CDC transport, stream processing, cross-Layer communication, audit logging, metrics collection, and command queues — avoiding the operational burden of multiple messaging systems.
Partitioning strategy determines scalability: Partitioning by entity ID guarantees message ordering; partitioning by tenant ID enables tenant isolation; partitioning by time optimizes historical data queries. Partition design is one of the most critical decisions in any Kafka architecture.
Exactly-once semantics require end-to-end guarantees: Kafka transactions alone are insufficient — you need full-chain idempotent design from producer to consumer. In coomia-dip, combining Flink checkpoints with Iceberg's optimistic concurrency achieves true end-to-end exactly-once delivery.

#Next Article

S8-06: Flink CDC 10 Best Practices — From Debezium Connector tuning to Flink CDC 3.0 new features, summarizing 10 production-grade best practices for real-time CDC data integration.

Tags: #apache-kafka #event-bus #cdc #stream-processing #cross-Layer #audit-log #coomia-dip #messaging