kafka

What this skill does

# Apache Kafka Core Knowledge

> **Deep Knowledge**: Use `mcp__documentation__fetch_docs` with technology: `kafka` for comprehensive documentation.

## Quick Start (Docker)

```yaml
# docker-compose.yml
services:
  kafka:
    image: bitnami/kafka:latest
    ports:
      - "9092:9092"
    environment:
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
```

```bash
# Start
docker-compose up -d

# Create topic
docker exec kafka kafka-topics.sh --create --topic my-topic \
  --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
```

## Core Concepts

| Concept | Description |
|---------|-------------|
| **Topic** | Named stream of records, append-only log |
| **Partition** | Ordered, immutable sequence within topic |
| **Offset** | Unique ID for record within partition |
| **Consumer Group** | Set of consumers sharing topic consumption |
| **Broker** | Kafka server handling storage and requests |
| **Replication Factor** | Number of partition copies across brokers |

## Architecture

```
┌─────────────────────────────────────────────────────────┐
│                    Kafka Cluster                        │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐            │
│  │Broker 0 │    │Broker 1 │    │Broker 2 │            │
│  │ P0(L)   │    │ P0(F)   │    │ P1(L)   │            │
│  │ P1(F)   │    │ P1(F)   │    │ P0(F)   │            │
│  └─────────┘    └─────────┘    └─────────┘            │
└─────────────────────────────────────────────────────────┘
         ▲                              │
         │                              ▼
   ┌──────────┐                  ┌──────────────┐
   │ Producer │                  │Consumer Group│
   └──────────┘                  │  C1  C2  C3  │
                                 └──────────────┘
```

## Producer Patterns

### Node.js (kafkajs)
```typescript
import { Kafka, Partitioners } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092'],
});

const producer = kafka.producer({
  createPartitioner: Partitioners.DefaultPartitioner,
  idempotent: true, // Enable exactly-once
});

await producer.connect();

// Send single message
await producer.send({
  topic: 'orders',
  messages: [
    {
      key: orderId,           // Partition key
      value: JSON.stringify(order),
      headers: {
        'correlation-id': correlationId,
        'source': 'order-service',
      },
    },
  ],
});

// Batch send
await producer.sendBatch({
  topicMessages: [
    {
      topic: 'orders',
      messages: orders.map(o => ({
        key: o.id,
        value: JSON.stringify(o),
      })),
    },
  ],
});

await producer.disconnect();
```

### Java (Spring Kafka)
```java
@Configuration
public class KafkaConfig {
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

@Service
public class OrderProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendOrder(Order order) {
        kafkaTemplate.send("orders", order.getId(), objectMapper.writeValueAsString(order))
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to send order", ex);
                }
            });
    }
}
```

### Python (confluent-kafka)
```python
from confluent_kafka import Producer
import json

conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'my-app',
    'acks': 'all',
    'enable.idempotence': True,
}

producer = Producer(conf)

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

# Send message
producer.produce(
    topic='orders',
    key=order_id.encode('utf-8'),
    value=json.dumps(order).encode('utf-8'),
    callback=delivery_callback,
    headers={'correlation-id': correlation_id}
)

producer.flush()  # Wait for delivery
```

### Go (segmentio/kafka-go)
```go
package main

import (
    "context"
    "encoding/json"
    "github.com/segmentio/kafka-go"
)

func main() {
    writer := &kafka.Writer{
        Addr:         kafka.TCP("localhost:9092"),
        Topic:        "orders",
        Balancer:     &kafka.LeastBytes{},
        RequiredAcks: kafka.RequireAll,
    }

    defer writer.Close()

    order := Order{ID: "123", Amount: 100}
    value, _ := json.Marshal(order)

    err := writer.WriteMessages(context.Background(),
        kafka.Message{
            Key:   []byte(order.ID),
            Value: value,
            Headers: []kafka.Header{
                {Key: "correlation-id", Value: []byte("abc123")},
            },
        },
    )
}
```

## Consumer Patterns

### Node.js (kafkajs)
```typescript
const consumer = kafka.consumer({
  groupId: 'order-processor',
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
});

await consumer.connect();
await consumer.subscribe({ topics: ['orders'], fromBeginning: false });

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const order = JSON.parse(message.value.toString());
    const correlationId = message.headers['correlation-id']?.toString();

    try {
      await processOrder(order);
      // Auto-commit on success
    } catch (error) {
      // Handle error - message will be redelivered
      throw error;
    }
  },
});

// Manual commit
await consumer.run({
  autoCommit: false,
  eachBatch: async ({ batch, resolveOffset, commitOffsetsIfNecessary }) => {
    for (const message of batch.messages) {
      await processMessage(message);
      resolveOffset(message.offset);
    }
    await commitOffsetsIfNecessary();
  },
});
```

### Java (Spring Kafka)
```java
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(AckMode.MANUAL);
        return factory;
    }
}

@Service
public class OrderConsumer {
    @KafkaListener(topics = "orders", groupId = "order-processor")
    public void consume(
            @Payload String message,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            @Header("correlation-id") String correlationId,
            Acknowledgment ack) {

        Order order = objectMap