┌─────────────────────────────────────────────────────────────────────┐
│                        Kafka Cluster                                │
│                                                                     │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐             │
│  │  Broker 1    │  │  Broker 2    │  │  Broker 3    │             │
│  │  (Controller)│  │              │  │              │             │
│  │              │  │              │  │              │             │
│  │ ┌──────────┐│  │ ┌──────────┐│  │ ┌──────────┐│             │
│  │ │ Topic A  ││  │ │ Topic A  ││  │ │ Topic A  ││             │
│  │ │ P0 (L)   ││  │ │ P0 (F)   ││  │ │ P1 (F)   ││             │
│  │ │ P1 (L)   ││  │ │ P2 (L)   ││  │ │ P2 (F)   ││             │
│  │ └──────────┘│  │ └──────────┘│  │ └──────────┘│             │
│  │              │  │              │  │              │             │
│  │ ┌──────────┐│  │ ┌──────────┐│  │ ┌──────────┐│             │
│  │ │ Topic B  ││  │ │ Topic B  ││  │ │ Topic B  ││             │
│  │ │ P0 (F)   ││  │ │ P0 (L)   ││  │ │ P1 (L)   ││             │
│  │ │ P1 (F)   ││  │ │          ││  │ │          ││             │
│  │ └──────────┘│  │ └──────────┘│  │ └──────────┘│             │
│  └──────────────┘  └──────────────┘  └──────────────┘             │
│         L = Leader   F = Follower                                   │
│                                                                     │
│  ┌─────────────────────────────────────────┐                       │
│  │          KRaft Controller Quorum        │                       │
│  │     (replaces ZooKeeper in 3.3+)        │                       │
│  └─────────────────────────────────────────┘                       │
└─────────────────────────────────────────────────────────────────────┘
        │                    │                    │
   ┌────┴────┐          ┌───┴────┐          ┌───┴────┐
   │Producer │          │Consumer│          │Consumer│
   │  App    │          │Group A │          │Group B │
   └─────────┘          └────────┘          └────────┘

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class OrderProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                  "kafka-1:9092,kafka-2:9092,kafka-3:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                  StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                  StringSerializer.class.getName());

        // Reliability settings
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        // Performance settings
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 20);
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);

        try (KafkaProducer<String, String> producer =
                 new KafkaProducer<>(props)) {

            ProducerRecord<String, String> record =
                new ProducerRecord<>("orders", "order-123",
                    "{\"id\":\"123\",\"amount\":99.99}");

            // Async send with callback
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("Send failed: " + exception);
                } else {
                    System.out.printf("Sent to %s-%d offset %d%n",
                        metadata.topic(), metadata.partition(),
                        metadata.offset());
                }
            });
        }
    }
}

from confluent_kafka import Producer
import json

conf = {
    'bootstrap.servers': 'kafka-1:9092,kafka-2:9092,kafka-3:9092',
    'acks': 'all',
    'enable.idempotence': True,
    'compression.type': 'lz4',
    'batch.size': 32768,
    'linger.ms': 20,
    'retries': 2147483647,
}

producer = Producer(conf)

def delivery_callback(err, msg):
    if err:
        print(f'Message delivery failed: {err}')
    else:
        print(f'Delivered to {msg.topic()} [{msg.partition()}] @ {msg.offset()}')

# Send messages
for i in range(100):
    order = {'id': f'order-{i}', 'amount': 99.99 + i}
    producer.produce(
        topic='orders',
        key=f'order-{i}',
        value=json.dumps(order).encode('utf-8'),
        callback=delivery_callback
    )

# Wait for all messages to be delivered
producer.flush()

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class OrderConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                  "kafka-1:9092,kafka-2:9092,kafka-3:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                  StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                  StringDeserializer.class.getName());

        // Consumer configuration
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024);
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);

        try (KafkaConsumer<String, String> consumer =
                 new KafkaConsumer<>(props)) {

            consumer.subscribe(List.of("orders"));

            while (true) {
                ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(1000));

                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Key: %s, Value: %s, " +
                        "Partition: %d, Offset: %d%n",
                        record.key(), record.value(),
                        record.partition(), record.offset());

                    // Process the record
                    processOrder(record.value());
                }

                // Manual commit after processing
                consumer.commitSync();
            }
        }
    }

    private static void processOrder(String orderJson) {
        // Business logic here
    }
}

from confluent_kafka import Consumer
import json

conf = {
    'bootstrap.servers': 'kafka-1:9092,kafka-2:9092,kafka-3:9092',
    'group.id': 'order-processor',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,
    'max.poll.interval.ms': 300000,
}

consumer = Consumer(conf)
consumer.subscribe(['orders'])

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

        order = json.loads(msg.value().decode('utf-8'))
        print(f"Received: {order}")

        # Process and commit
        process_order(order)
        consumer.commit(asynchronous=False)

except KeyboardInterrupt:
    pass
finally:
    consumer.close()

{
  "name": "postgres-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "replicator",
    "database.password": "${vault:postgres/password}",
    "database.dbname": "orders_db",
    "database.server.name": "orders",
    "table.include.list": "public.orders,public.customers",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_orders",
    "publication.name": "dbz_publication",
    "topic.prefix": "cdc",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "cdc\\.public\\.(.*)",
    "transforms.route.replacement": "cdc-$1",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

{
  "name": "elasticsearch-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "topics": "cdc-orders",
    "connection.url": "http://elasticsearch:9200",
    "type.name": "_doc",
    "key.ignore": false,
    "schema.ignore": true,
    "behavior.on.null.values": "delete",
    "write.method": "upsert",
    "batch.size": 2000,
    "max.buffered.records": 20000,
    "flush.timeout.ms": 120000,
    "transforms": "extractKey,dropPrefix",
    "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.extractKey.field": "id",
    "transforms.dropPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.dropPrefix.regex": "cdc-(.*)",
    "transforms.dropPrefix.replacement": "$1"
  }
}

{
  "name": "s3-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "orders,customers",
    "s3.region": "eu-west-1",
    "s3.bucket.name": "data-lake-raw",
    "s3.part.size": 67108864,
    "flush.size": 10000,
    "rotate.interval.ms": 600000,
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "partition.duration.ms": 3600000,
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    "locale": "en-US",
    "timezone": "UTC",
    "timestamp.extractor": "RecordField",
    "timestamp.field": "created_at"
  }
}

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.common.serialization.Serdes;

import java.time.Duration;
import java.util.Properties;

public class OrderStreamProcessor {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,
                  "order-analytics");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
                  "kafka-1:9092,kafka-2:9092,kafka-3:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
                  Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
                  Serdes.String().getClass());
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
                  StreamsConfig.EXACTLY_ONCE_V2);

        StreamsBuilder builder = new StreamsBuilder();

        // Read from input topic
        KStream<String, String> orders = builder.stream("orders");

        // Filter high-value orders
        KStream<String, String> highValue = orders
            .filter((key, value) -> {
                Order order = parseOrder(value);
                return order.amount() > 1000;
            });

        // Branch into categories
        orders
            .split(Named.as("order-"))
            .branch((key, value) -> parseOrder(value).amount() > 1000,
                    Branched.as("high-value"))
            .branch((key, value) -> parseOrder(value).amount() > 100,
                    Branched.as("medium-value"))
            .defaultBranch(Branched.as("low-value"));

        // Windowed aggregation: orders per minute per region
        orders
            .groupBy((key, value) -> parseOrder(value).region())
            .windowedBy(TimeWindows.ofSizeWithNoGrace(
                Duration.ofMinutes(1)))
            .count(Materialized.as("orders-per-minute"))
            .toStream()
            .to("order-counts");

        // Join orders with customer data
        KTable<String, String> customers = builder.table("customers");

        orders
            .selectKey((key, value) -> parseOrder(value).customerId())
            .join(customers,
                  (order, customer) -> enrichOrder(order, customer))
            .to("enriched-orders");

        // Build and start
        KafkaStreams streams = new KafkaStreams(
            builder.build(), props);
        streams.start();

        Runtime.getRuntime().addShutdownHook(
            new Thread(streams::close));
    }
}

# server.properties — Performance tuning

# Network and I/O threads
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600

# Log configuration
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

# Replication
replica.fetch.max.bytes=10485760
replica.fetch.wait.max.ms=500
replica.socket.receive.buffer.bytes=1048576
num.replica.fetchers=4

# Compression
compression.type=producer

# Message sizes
message.max.bytes=10485760

# Messages produced per second per topic
sum(rate(kafka_server_brokertopicmetrics_messagesinpersec_count[5m])) by (topic)

# Consumer group lag
sum(kafka_consumergroup_lag) by (consumergroup, topic)

# Under-replicated partitions
kafka_server_replicamanager_underreplicatedpartitions

# Broker CPU usage
rate(process_cpu_seconds_total{job="kafka"}[5m])

# Request rate by type
sum(rate(kafka_network_requestmetrics_requestspersec_count[5m])) by (request)

┌──────────┐     ┌───────────┐     ┌──────────────┐
│ Command  │────>│  Kafka    │────>│ Event Store  │
│ Service  │     │  Topic    │     │ (Compacted)  │
└──────────┘     └───────────┘     └──────────────┘
                      │
                 ┌────┴────┐
                 │ Consumer│
                 │ (CQRS   │
                 │  Read)  │
                 └─────────┘

// Handle poison pills with DLQ
try {
    processRecord(record);
    consumer.commitSync();
} catch (DeserializationException e) {
    // Send to DLQ
    producer.send(new ProducerRecord<>(
        "orders.dlq",
        record.key(),
        record.value()
    ));
    consumer.commitSync();
}

-- Application writes to orders table and outbox table
-- in a single database transaction
BEGIN;

INSERT INTO orders (id, customer_id, amount)
VALUES ('order-123', 'cust-456', 99.99);

INSERT INTO outbox (id, aggregate_type, aggregate_id, event_type, payload)
VALUES (gen_random_uuid(), 'Order', 'order-123', 'OrderCreated',
        '{"id":"order-123","amount":99.99}');

COMMIT;

-- Debezium CDC connector reads the outbox table
-- and publishes events to Kafka