import { Kafka, logLevel, CompressionTypes } from 'kafkajs';

interface ProducerConfig {
  brokers: string[];
  clientId: string;
  idempotent?: boolean;
  maxInFlightRequests?: number;
  compression?: CompressionTypes;
}

class KafkaProducer {
  private kafka: Kafka;
  private producer: any;

  constructor(config: ProducerConfig) {
    this.kafka = new Kafka({
      clientId: config.clientId,
      brokers: config.brokers,
      logLevel: logLevel.INFO,
      ssl: true,
      sasl: {
        mechanism: 'plain',
        username: process.env.KAFKA_USER || '',
        password: process.env.KAFKA_PASSWORD || '',
      },
    });

    this.producer = this.kafka.producer({
      idempotent: config.idempotent ?? true,
      maxInFlightRequests: config.maxInFlightRequests ?? 5,
      compression: config.compression ?? CompressionTypes.Snappy,
      // Tuning for high throughput
      requestTimeout: 30000,
      retry: {
        initialRetryTime: 100,
        retries: 8,
        randomizationFactor: 0.2,
      },
    });
  }

  async connect(): Promise<void> {
    await this.producer.connect();
  }

  async disconnect(): Promise<void> {
    await this.producer.disconnect();
  }

  // Single event with key-based partitioning
  async sendEvent(topic: string, key: string, value: any): Promise<void> {
    await this.producer.send({
      topic,
      messages: [
        {
          key,
          value: JSON.stringify(value),
          timestamp: Date.now().toString(),
        },
      ],
    });
  }

  // Batch send for higher throughput
  async sendBatch(topic: string, events: Array<{ key: string; value: any }>): Promise<void> {
    await this.producer.send({
      topic,
      messages: events.map(({ key, value }) => ({
        key,
        value: JSON.stringify(value),
        timestamp: Date.now().toString(),
      })),
      // Batch settings for throughput
      timeout: 30000,
      compression: CompressionTypes.Snappy,
    });
  }

  // Transactional writes (exactly-once)
  async sendTransactional(
    topic: string,
    events: Array<{ key: string; value: any }>
  ): Promise<void> {
    const transaction = await this.producer.transaction();
    try {
      await transaction.send({
        topic,
        messages: events.map(({ key, value }) => ({
          key,
          value: JSON.stringify(value),
        })),
      });
      await transaction.commit();
    } catch (error) {
      await transaction.abort();
      throw error;
    }
  }
}

// Usage example
const producer = new KafkaProducer({
  brokers: ['kafka-1:9092', 'kafka-2:9092', 'kafka-3:9092'],
  clientId: 'order-service',
  idempotent: true,
  compression: CompressionTypes.Snappy,
});

await producer.connect();

// Send order event
await producer.sendEvent('orders', 'order-123', {
  orderId: 'order-123',
  customerId: 'cust-456',
  amount: 99.99,
  timestamp: new Date().toISOString(),
});

await producer.disconnect();

// pom.xml dependency: org.springframework.kafka:spring-kafka
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.core.*;
import org.springframework.kafka.support.SendResult;

import java.util.*;
import java.util.concurrent.CompletableFuture;

public class KafkaProducerService {
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper = new ObjectMapper();

    public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    // Single event with key-based partitioning
    public CompletableFuture<SendResult<String, String>> sendEvent(
            String topic, String key, Object value) throws Exception {
        String json = objectMapper.writeValueAsString(value);
        return kafkaTemplate.send(topic, key, json);
    }

    // Batch send for higher throughput
    public void sendBatch(String topic, List<Map<String, Object>> events) throws Exception {
        for (Map<String, Object> event : events) {
            String key = (String) event.get("key");
            String json = objectMapper.writeValueAsString(event.get("value"));
            kafkaTemplate.send(topic, key, json);
        }
        kafkaTemplate.flush();
    }

    // Transactional writes (exactly-once)
    public void sendTransactional(String topic, List<Map<String, Object>> events) throws Exception {
        kafkaTemplate.executeInTransaction(kt -> {
            for (Map<String, Object> event : events) {
                try {
                    String key = (String) event.get("key");
                    String json = objectMapper.writeValueAsString(event.get("value"));
                    kt.send(topic, key, json);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
            return true;
        });
    }
}

// Spring Boot configuration
@Configuration
public class KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092,kafka-2:9092,kafka-3:9092");
        config.put(ProducerConfig.CLIENT_ID_CONFIG, "order-service");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        config.put(ProducerConfig.RETRIES_CONFIG, 8);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(config);
    }

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

# pip install confluent-kafka aiokafka
import asyncio
import json
import os
from dataclasses import dataclass
from typing import Any
from aiokafka import AIOKafkaProducer
from aiokafka.helpers import create_ssl_context

@dataclass
class ProducerConfig:
    brokers: list[str]
    client_id: str
    idempotent: bool = True
    compression: str = "snappy"

class KafkaProducerService:
    def __init__(self, config: ProducerConfig):
        self.config = config
        self.producer: AIOKafkaProducer | None = None

    async def connect(self) -> None:
        ssl_context = create_ssl_context()
        self.producer = AIOKafkaProducer(
            bootstrap_servers=",".join(self.config.brokers),
            client_id=self.config.client_id,
            enable_idempotence=self.config.idempotent,
            compression_type=self.config.compression,
            max_in_flight_requests_per_connection=5,
            request_timeout_ms=30000,
            retry_backoff_ms=100,
            ssl_context=ssl_context,
            sasl_mechanism="PLAIN",
            sasl_plain_username=os.getenv("KAFKA_USER", ""),
            sasl_plain_password=os.getenv("KAFKA_PASSWORD", ""),
        )
        await self.producer.start()

    async def disconnect(self) -> None:
        if self.producer:
            await self.producer.stop()

    # Single event with key-based partitioning
    async def send_event(self, topic: str, key: str, value: Any) -> None:
        await self.producer.send(
            topic,
            key=key.encode(),
            value=json.dumps(value).encode(),
            timestamp_ms=int(asyncio.get_event_loop().time() * 1000),
        )

    # Batch send for higher throughput
    async def send_batch(self, topic: str, events: list[dict]) -> None:
        batch = self.producer.create_batch()
        for event in events:
            batch.append(
                key=event["key"].encode(),
                value=json.dumps(event["value"]).encode(),
                timestamp=None,
            )
        partitions = await self.producer.partitions_for(topic)
        partition = next(iter(partitions))
        await self.producer.send_batch(batch, topic, partition=partition)

    # Transactional writes (exactly-once)
    async def send_transactional(self, topic: str, events: list[dict]) -> None:
        async with self.producer.transaction():
            for event in events:
                await self.producer.send(
                    topic,
                    key=event["key"].encode(),
                    value=json.dumps(event["value"]).encode(),
                )


# Usage example
async def main():
    producer = KafkaProducerService(ProducerConfig(
        brokers=["kafka-1:9092", "kafka-2:9092", "kafka-3:9092"],
        client_id="order-service",
        idempotent=True,
        compression="snappy",
    ))
    await producer.connect()
    await producer.send_event("orders", "order-123", {
        "orderId": "order-123",
        "customerId": "cust-456",
        "amount": 99.99,
        "timestamp": "2024-01-01T00:00:00Z",
    })
    await producer.disconnect()

// NuGet: Confluent.Kafka
using Confluent.Kafka;
using System.Text.Json;

public class ProducerConfig
{
    public string[] Brokers { get; set; } = [];
    public string ClientId { get; set; } = "";
    public bool Idempotent { get; set; } = true;
    public CompressionType Compression { get; set; } = CompressionType.Snappy;
}

public class KafkaProducerService : IDisposable
{
    private readonly IProducer<string, string> _producer;

    public KafkaProducerService(ProducerConfig config)
    {
        var producerConfig = new Confluent.Kafka.ProducerConfig
        {
            BootstrapServers = string.Join(",", config.Brokers),
            ClientId = config.ClientId,
            EnableIdempotence = config.Idempotent,
            CompressionType = config.Compression,
            MaxInFlight = 5,
            MessageTimeoutMs = 30000,
            Acks = Acks.All,
            SaslMechanism = SaslMechanism.Plain,
            SaslUsername = Environment.GetEnvironmentVariable("KAFKA_USER") ?? "",
            SaslPassword = Environment.GetEnvironmentVariable("KAFKA_PASSWORD") ?? "",
            SecurityProtocol = SecurityProtocol.SaslSsl,
        };
        _producer = new ProducerBuilder<string, string>(producerConfig).Build();
    }

    // Single event with key-based partitioning
    public async Task SendEventAsync(string topic, string key, object value)
    {
        var json = JsonSerializer.Serialize(value);
        await _producer.ProduceAsync(topic, new Message<string, string>
        {
            Key = key,
            Value = json,
            Timestamp = new Timestamp(DateTime.UtcNow),
        });
    }

    // Batch send for higher throughput
    public async Task SendBatchAsync(string topic, IEnumerable<(string Key, object Value)> events)
    {
        foreach (var (key, value) in events)
        {
            var json = JsonSerializer.Serialize(value);
            _producer.Produce(topic, new Message<string, string> { Key = key, Value = json });
        }
        _producer.Flush(TimeSpan.FromSeconds(30));
        await Task.CompletedTask;
    }

    // Transactional writes (exactly-once)
    public async Task SendTransactionalAsync(string topic, IEnumerable<(string Key, object Value)> events)
    {
        _producer.InitTransactions(TimeSpan.FromSeconds(30));
        _producer.BeginTransaction();
        try
        {
            foreach (var (key, value) in events)
            {
                var json = JsonSerializer.Serialize(value);
                _producer.Produce(topic, new Message<string, string> { Key = key, Value = json });
            }
            _producer.CommitTransaction();
        }
        catch
        {
            _producer.AbortTransaction();
            throw;
        }
        await Task.CompletedTask;
    }

    public void Dispose() => _producer.Dispose();
}

// Strategy 1: Hash-based (default) — key is hashed to partition
// Same key always goes to same partition (ordered within key)
await producer.sendEvent('orders', 'customer-123', orderData);

// Strategy 2: Round-robin (null key)
// Distributes events evenly across partitions, but no ordering
await producer.sendEvent('events', null, eventData);

// Strategy 3: Custom strategy — use business domain
// Use customer ID to co-locate related events
const customerId = order.customerId;
await producer.sendEvent('orders', customerId, orderData);

// Strategy 1: Hash-based — same key always goes to same partition
producerService.sendEvent("orders", "customer-123", orderData).get();

// Strategy 2: Round-robin (null key)
producerService.sendEvent("events", null, eventData).get();

// Strategy 3: Custom strategy — use business domain
String customerId = order.getCustomerId();
producerService.sendEvent("orders", customerId, orderData).get();

# Strategy 1: Hash-based — same key always goes to same partition
await producer.send_event("orders", "customer-123", order_data)

# Strategy 2: Round-robin (null key) — distribute evenly
await producer.producer.send("events", value=json.dumps(event_data).encode())

# Strategy 3: Custom strategy — use business domain
customer_id = order["customerId"]
await producer.send_event("orders", customer_id, order_data)

// Strategy 1: Hash-based — same key always goes to same partition
await producerService.SendEventAsync("orders", "customer-123", orderData);

// Strategy 2: Round-robin (null key)
await producerService.SendEventAsync("events", null!, eventData);

// Strategy 3: Custom strategy — use business domain
var customerId = order.CustomerId;
await producerService.SendEventAsync("orders", customerId, orderData);

// Batch events together
const events: Array<{ key: string; value: any }> = [];

for (let i = 0; i < 1000; i++) {
  events.push({
    key: `customer-${i % 100}`,
    value: { eventId: i, timestamp: Date.now() },
  });
}

// Send batch with compression
await producer.sendBatch('events', events);

List<Map<String, Object>> events = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
    events.add(Map.of(
        "key", "customer-" + (i % 100),
        "value", Map.of("eventId", i, "timestamp", System.currentTimeMillis())
    ));
}
// Send batch with compression
producerService.sendBatch("events", events);

events = [
    {
        "key": f"customer-{i % 100}",
        "value": {"eventId": i, "timestamp": int(asyncio.get_event_loop().time() * 1000)},
    }
    for i in range(1000)
]
# Send batch with compression
await producer.send_batch("events", events)

var events = Enumerable.Range(0, 1000).Select(i => (
    Key: $"customer-{i % 100}",
    Value: (object)new { eventId = i, timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() }
));
// Send batch with compression
await producerService.SendBatchAsync("events", events);

interface ConsumerConfig {
  brokers: string[];
  groupId: string;
  clientId: string;
  topics: string[];
}

class KafkaConsumer {
  private kafka: Kafka;
  private consumer: any;
  private groupId: string;

  constructor(config: ConsumerConfig) {
    this.kafka = new Kafka({
      clientId: config.clientId,
      brokers: config.brokers,
      logLevel: logLevel.INFO,
      ssl: true,
      sasl: {
        mechanism: 'plain',
        username: process.env.KAFKA_USER || '',
        password: process.env.KAFKA_PASSWORD || '',
      },
    });

    this.groupId = config.groupId;
    this.consumer = this.kafka.consumer({
      groupId: config.groupId,
      sessionTimeout: 30000,
      heartbeatInterval: 3000,
      maxBytesPerPartition: 1048576, // 1MB
      // Offset commit strategy
      allowAutoTopicCreation: false,
    });
  }

  async connect(): Promise<void> {
    await this.consumer.connect();
  }

  async subscribe(topics: string[]): Promise<void> {
    await this.consumer.subscribe({ topics, fromBeginning: false });
  }

  async run(
    onMessage: (topic: string, partition: number, message: any) => Promise<void>
  ): Promise<void> {
    await this.consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        try {
          const value = JSON.parse(message.value?.toString() || '{}');
          await onMessage(topic, partition, value);
          // Offset committed automatically with autoCommit
        } catch (error) {
          console.error(`Error processing message from ${topic}:${partition}`, error);
          // Handle poison pill messages appropriately
          throw error;
        }
      },
      // Manual offset management for critical applications
      autoCommit: {
        interval: 5000,
        threshold: 100,
      },
    });
  }

  async disconnect(): Promise<void> {
    await this.consumer.disconnect();
  }

  // Reset offset for replaying events
  async resetOffset(topic: string, offset: string): Promise<void> {
    await this.consumer.seek({ topic, partition: 0, offset });
  }
}

// Usage example
const consumer = new KafkaConsumer({
  brokers: ['kafka-1:9092', 'kafka-2:9092', 'kafka-3:9092'],
  groupId: 'order-processing-service',
  clientId: 'order-processor-1',
  topics: ['orders'],
});

await consumer.connect();
await consumer.subscribe(['orders']);

await consumer.run(async (topic, partition, message) => {
  console.log(`Processing order from partition ${partition}:`, message);

  // Process order (call external API, update database, etc.)
  await processOrder(message);

  // Offset is committed automatically by eachMessage
});

// Spring Kafka @KafkaListener approach
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.core.ConsumerFactory;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

@Service
public class KafkaConsumerService {

    @KafkaListener(
        topics = "orders",
        groupId = "order-processing-service",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            var value = objectMapper.readValue(record.value(), Map.class);
            System.out.printf("Processing order from partition %d: %s%n",
                record.partition(), value);
            processOrder(value);
            ack.acknowledge(); // Manual commit after successful processing
        } catch (Exception e) {
            log.error("Error processing message from {}:{}", record.topic(), record.partition(), e);
            throw new RuntimeException(e);
        }
    }
}

@Configuration
public class KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092,kafka-2:9092,kafka-3:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-service");
        config.put(ConsumerConfig.CLIENT_ID_CONFIG, "order-processor-1");
        config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        config.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
        config.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1048576);
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(config);
    }

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

# pip install aiokafka
from aiokafka import AIOKafkaConsumer
import asyncio
import json
import os

class KafkaConsumerService:
    def __init__(self, brokers: list[str], group_id: str, client_id: str):
        self.brokers = brokers
        self.group_id = group_id
        self.client_id = client_id
        self.consumer: AIOKafkaConsumer | None = None

    async def connect(self) -> None:
        self.consumer = AIOKafkaConsumer(
            bootstrap_servers=",".join(self.brokers),
            group_id=self.group_id,
            client_id=self.client_id,
            session_timeout_ms=30000,
            heartbeat_interval_ms=3000,
            max_partition_fetch_bytes=1048576,
            enable_auto_commit=True,
            auto_commit_interval_ms=5000,
            sasl_mechanism="PLAIN",
            sasl_plain_username=os.getenv("KAFKA_USER", ""),
            sasl_plain_password=os.getenv("KAFKA_PASSWORD", ""),
        )
        await self.consumer.start()

    async def subscribe(self, topics: list[str]) -> None:
        self.consumer.subscribe(topics)

    async def run(self, on_message) -> None:
        async for msg in self.consumer:
            try:
                value = json.loads(msg.value.decode())
                await on_message(msg.topic, msg.partition, value)
            except Exception as e:
                print(f"Error processing message from {msg.topic}:{msg.partition}: {e}")
                raise

    async def disconnect(self) -> None:
        if self.consumer:
            await self.consumer.stop()

    # Reset offset for replaying events
    async def reset_offset(self, topic: str, partition: int, offset: int) -> None:
        from aiokafka import TopicPartition
        tp = TopicPartition(topic, partition)
        self.consumer.seek(tp, offset)


# Usage example
async def main():
    consumer = KafkaConsumerService(
        brokers=["kafka-1:9092", "kafka-2:9092", "kafka-3:9092"],
        group_id="order-processing-service",
        client_id="order-processor-1",
    )
    await consumer.connect()
    await consumer.subscribe(["orders"])

    async def handle_message(topic, partition, message):
        print(f"Processing order from partition {partition}: {message}")
        await process_order(message)

    await consumer.run(handle_message)

// NuGet: Confluent.Kafka
using Confluent.Kafka;
using System.Text.Json;

public class KafkaConsumerService : BackgroundService
{
    private readonly IConsumer<string, string> _consumer;
    private readonly ILogger<KafkaConsumerService> _logger;

    public KafkaConsumerService(ILogger<KafkaConsumerService> logger)
    {
        _logger = logger;
        var config = new ConsumerConfig
        {
            BootstrapServers = "kafka-1:9092,kafka-2:9092,kafka-3:9092",
            GroupId = "order-processing-service",
            ClientId = "order-processor-1",
            SessionTimeoutMs = 30000,
            HeartbeatIntervalMs = 3000,
            MaxPartitionFetchBytes = 1048576,
            EnableAutoCommit = true,
            AutoCommitIntervalMs = 5000,
            SaslMechanism = SaslMechanism.Plain,
            SaslUsername = Environment.GetEnvironmentVariable("KAFKA_USER") ?? "",
            SaslPassword = Environment.GetEnvironmentVariable("KAFKA_PASSWORD") ?? "",
            SecurityProtocol = SecurityProtocol.SaslSsl,
            AutoOffsetReset = AutoOffsetReset.Latest,
        };
        _consumer = new ConsumerBuilder<string, string>(config).Build();
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _consumer.Subscribe("orders");

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var result = _consumer.Consume(stoppingToken);
                var value = JsonSerializer.Deserialize<Dictionary<string, object>>(result.Message.Value);
                _logger.LogInformation(
                    "Processing order from partition {Partition}: {Message}",
                    result.Partition.Value, value);
                await ProcessOrderAsync(value!);
            }
            catch (OperationCanceledException) { break; }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing message");
                throw;
            }
        }
        _consumer.Close();
    }

    // Reset offset for replaying events
    public void ResetOffset(string topic, int partition, long offset)
    {
        _consumer.Seek(new TopicPartitionOffset(topic, partition, offset));
    }

    private Task ProcessOrderAsync(Dictionary<string, object> message) =>
        Task.CompletedTask; // Replace with real logic
}

// 1. Auto-commit (default) — simple but risky
// Risk: lose progress if consumer crashes between commit and processing

// 2. Manual commit after processing
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    try {
      await processMessage(message);
      // Commit only after successful processing
      await consumer.commitOffsets([
        {
          topic,
          partition,
          offset: (parseInt(message.offset) + 1).toString(),
        },
      ]);
    } catch (error) {
      // Don't commit, consumer will retry from this offset
      throw error;
    }
  },
  autoCommit: false,
});

// 3. Batch commit — balance throughput and safety
// Commit after processing N messages
let processedCount = 0;
const BATCH_SIZE = 100;

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    await processMessage(message);
    processedCount++;

    if (processedCount >= BATCH_SIZE) {
      await consumer.commitOffsets([
        {
          topic,
          partition,
          offset: (parseInt(message.offset) + 1).toString(),
        },
      ]);
      processedCount = 0;
    }
  },
  autoCommit: false,
});

// 1. Auto-commit — set enable.auto.commit=true in config (default)

// 2. Manual commit after processing (AckMode.MANUAL)
@KafkaListener(topics = "orders", groupId = "order-processors")
public void consumeManual(ConsumerRecord<String, String> record, Acknowledgment ack) {
    try {
        processMessage(record.value());
        ack.acknowledge(); // Commit only after successful processing
    } catch (Exception e) {
        // Don't ack — consumer will retry from this offset
        throw new RuntimeException(e);
    }
}

// 3. Batch commit — AckMode.COUNT
// In container factory: factory.getContainerProperties().setAckCount(100);
@KafkaListener(topics = "orders", groupId = "order-processors",
    containerFactory = "batchListenerContainerFactory")
public void consumeBatch(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
    for (var record : records) {
        processMessage(record.value());
    }
    ack.acknowledge(); // Commit after entire batch
}

from aiokafka import AIOKafkaConsumer, TopicPartition

# 2. Manual commit after processing
consumer = AIOKafkaConsumer(
    "orders",
    bootstrap_servers="kafka-1:9092",
    group_id="order-processors",
    enable_auto_commit=False,
)
await consumer.start()

async for msg in consumer:
    try:
        await process_message(json.loads(msg.value.decode()))
        # Commit only after successful processing
        await consumer.commit({
            TopicPartition(msg.topic, msg.partition): msg.offset + 1
        })
    except Exception as e:
        # Don't commit — consumer will retry from this offset
        raise

# 3. Batch commit — process N messages then commit
processed_count = 0
BATCH_SIZE = 100
last_offsets = {}

async for msg in consumer:
    await process_message(json.loads(msg.value.decode()))
    processed_count += 1
    last_offsets[TopicPartition(msg.topic, msg.partition)] = msg.offset + 1

    if processed_count >= BATCH_SIZE:
        await consumer.commit(last_offsets)
        processed_count = 0
        last_offsets = {}

// 2. Manual commit after processing
var config = new ConsumerConfig
{
    EnableAutoCommit = false,
    // ...other config
};
using var consumer = new ConsumerBuilder<string, string>(config).Build();
consumer.Subscribe("orders");

while (true)
{
    var result = consumer.Consume();
    try
    {
        await ProcessMessageAsync(result.Message.Value);
        // Commit only after successful processing
        consumer.Commit(result);
    }
    catch (Exception)
    {
        // Don't commit — consumer will retry from this offset
        throw;
    }
}

// 3. Batch commit — commit after N messages
int processedCount = 0;
const int BATCH_SIZE = 100;
ConsumeResult<string, string>? lastResult = null;

while (true)
{
    var result = consumer.Consume();
    await ProcessMessageAsync(result.Message.Value);
    processedCount++;
    lastResult = result;

    if (processedCount >= BATCH_SIZE && lastResult != null)
    {
        consumer.Commit(lastResult);
        processedCount = 0;
        lastResult = null;
    }
}

// Configuration for at-least-once
const producer = kafka.producer({
  acks: 1, // Wait for leader acknowledgment
  retries: 5,
  timeout: 30000,
});

// Consumer: commit offset after processing
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    await processMessage(message); // Must be idempotent!
    await consumer.commitOffsets([...]);
  },
});

// Configuration for at-least-once
@Bean
public ProducerFactory<String, String> atLeastOnceProducerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.ACKS_CONFIG, "1"); // Leader acknowledgment
    config.put(ProducerConfig.RETRIES_CONFIG, 5);
    config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
    return new DefaultKafkaProducerFactory<>(config);
}

// Consumer: commit offset after processing (AckMode.MANUAL)
@KafkaListener(topics = "orders", groupId = "processors")
public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
    processMessage(record.value()); // Must be idempotent!
    ack.acknowledge();
}

# Configuration for at-least-once
producer = AIOKafkaProducer(
    bootstrap_servers="kafka-1:9092",
    acks=1,  # Wait for leader acknowledgment
    request_timeout_ms=30000,
)

# Consumer: commit offset after processing
async for msg in consumer:
    await process_message(json.loads(msg.value.decode()))  # Must be idempotent!
    await consumer.commit({TopicPartition(msg.topic, msg.partition): msg.offset + 1})

// Configuration for at-least-once
var producerConfig = new Confluent.Kafka.ProducerConfig
{
    Acks = Acks.Leader, // Wait for leader acknowledgment
    MessageTimeoutMs = 30000,
    // Retries handled by EnableIdempotence or manual retry logic
};

// Consumer: commit offset after processing
while (true)
{
    var result = consumer.Consume();
    await ProcessMessageAsync(result.Message.Value); // Must be idempotent!
    consumer.Commit(result);
}

// Idempotent processing example
async function processOrderIdempotent(order: any) {
  // Check if we've already processed this order
  const exists = await database.query(
    'SELECT id FROM processed_orders WHERE order_id = ?',
    [order.orderId]
  );

  if (exists) {
    console.log('Order already processed, skipping');
    return;
  }

  // Process order
  await database.execute('INSERT INTO orders VALUES (?)', [order]);
  await database.execute('INSERT INTO processed_orders VALUES (?, ?)', [
    order.orderId,
    Date.now(),
  ]);
}

@Service
public class IdempotentOrderService {
    private final JdbcTemplate jdbcTemplate;

    public void processOrderIdempotent(Map<String, Object> order) {
        String orderId = (String) order.get("orderId");

        // Check if we've already processed this order
        Integer count = jdbcTemplate.queryForObject(
            "SELECT COUNT(*) FROM processed_orders WHERE order_id = ?",
            Integer.class, orderId);

        if (count != null && count > 0) {
            log.info("Order already processed, skipping: {}", orderId);
            return;
        }

        // Process order
        jdbcTemplate.update("INSERT INTO orders VALUES (?)", orderId);
        jdbcTemplate.update(
            "INSERT INTO processed_orders VALUES (?, ?)",
            orderId, System.currentTimeMillis());
    }
}

async def process_order_idempotent(order: dict, db) -> None:
    order_id = order["orderId"]

    # Check if we've already processed this order
    result = await db.fetch_one(
        "SELECT id FROM processed_orders WHERE order_id = :order_id",
        {"order_id": order_id},
    )

    if result:
        print(f"Order already processed, skipping: {order_id}")
        return

    # Process order
    await db.execute("INSERT INTO orders VALUES (:order)", {"order": order_id})
    await db.execute(
        "INSERT INTO processed_orders VALUES (:order_id, :ts)",
        {"order_id": order_id, "ts": int(asyncio.get_event_loop().time() * 1000)},
    )

public class IdempotentOrderService
{
    private readonly IDbConnection _db;

    public async Task ProcessOrderIdempotentAsync(Dictionary<string, object> order)
    {
        var orderId = order["orderId"].ToString()!;

        // Check if we've already processed this order
        var exists = await _db.QueryFirstOrDefaultAsync<int>(
            "SELECT COUNT(1) FROM processed_orders WHERE order_id = @OrderId",
            new { OrderId = orderId });

        if (exists > 0)
        {
            Console.WriteLine($"Order already processed, skipping: {orderId}");
            return;
        }

        // Process order
        await _db.ExecuteAsync("INSERT INTO orders VALUES (@OrderId)", new { OrderId = orderId });
        await _db.ExecuteAsync(
            "INSERT INTO processed_orders VALUES (@OrderId, @Ts)",
            new { OrderId = orderId, Ts = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() });
    }
}

// Configuration for at-most-once
const producer = kafka.producer({
  acks: 0, // No acknowledgment, fire and forget
  retries: 0,
});

// Consumer: commit offset before processing
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    await consumer.commitOffsets([...]); // Commit first
    await processMessage(message); // Process after
  },
});

// Configuration for at-most-once
@Bean
public ProducerFactory<String, String> atMostOnceProducerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.ACKS_CONFIG, "0"); // No acknowledgment
    config.put(ProducerConfig.RETRIES_CONFIG, 0);
    return new DefaultKafkaProducerFactory<>(config);
}

// Consumer: commit offset before processing
@KafkaListener(topics = "orders", groupId = "processors")
public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
    ack.acknowledge(); // Commit first
    processMessage(record.value()); // Process after
}

# Configuration for at-most-once
producer = AIOKafkaProducer(
    bootstrap_servers="kafka-1:9092",
    acks=0,  # No acknowledgment, fire and forget
)

# Consumer: commit offset before processing
async for msg in consumer:
    await consumer.commit({  # Commit first
        TopicPartition(msg.topic, msg.partition): msg.offset + 1
    })
    await process_message(json.loads(msg.value.decode()))  # Process after

// Configuration for at-most-once
var producerConfig = new Confluent.Kafka.ProducerConfig
{
    Acks = Acks.None, // No acknowledgment, fire and forget
};

// Consumer: commit offset before processing
while (true)
{
    var result = consumer.Consume();
    consumer.Commit(result); // Commit first
    await ProcessMessageAsync(result.Message.Value); // Process after
}

// Producer configuration for exactly-once
const producer = kafka.producer({
  idempotent: true, // Enable idempotent producer
  maxInFlightRequests: 5, // Maintain ordering
  transactionTimeout: 60000,
});

// Transactional write
const transaction = await producer.transaction();
try {
  await transaction.send({
    topic: 'orders',
    messages: [
      { key: orderId, value: JSON.stringify(orderData) },
    ],
  });
  await transaction.commit();
} catch (error) {
  await transaction.abort();
  throw error;
}

// Consumer: enable isolation level
const consumer = kafka.consumer({
  groupId: 'order-processors',
  isolationLevel: 1, // Read committed — only committed messages
});

// Producer configuration for exactly-once
@Bean
public ProducerFactory<String, String> exactlyOnceProducerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
    config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-1");
    config.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60000);
    return new DefaultKafkaProducerFactory<>(config);
}

// Transactional write
kafkaTemplate.executeInTransaction(kt -> {
    kt.send("orders", orderId, objectMapper.writeValueAsString(orderData));
    return true;
});

// Consumer: read committed isolation level
config.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

# Producer configuration for exactly-once
producer = AIOKafkaProducer(
    bootstrap_servers="kafka-1:9092",
    enable_idempotence=True,
    max_in_flight_requests_per_connection=5,
    transactional_id="order-tx-1",
    transaction_timeout_ms=60000,
)
await producer.start()

# Transactional write
async with producer.transaction():
    await producer.send("orders", key=order_id.encode(),
                        value=json.dumps(order_data).encode())

# Consumer: read committed isolation level
consumer = AIOKafkaConsumer(
    "orders",
    bootstrap_servers="kafka-1:9092",
    group_id="order-processors",
    isolation_level="read_committed",
)

// Producer configuration for exactly-once
var producerConfig = new Confluent.Kafka.ProducerConfig
{
    EnableIdempotence = true,
    MaxInFlight = 5,
    TransactionalId = "order-tx-1",
    TransactionTimeoutMs = 60000,
};
var producer = new ProducerBuilder<string, string>(producerConfig).Build();

// Transactional write
producer.InitTransactions(TimeSpan.FromSeconds(30));
producer.BeginTransaction();
try
{
    producer.Produce("orders", new Message<string, string>
    {
        Key = orderId,
        Value = JsonSerializer.Serialize(orderData),
    });
    producer.CommitTransaction();
}
catch
{
    producer.AbortTransaction();
    throw;
}

// Consumer: read committed isolation level
var consumerConfig = new ConsumerConfig
{
    IsolationLevel = IsolationLevel.ReadCommitted,
    GroupId = "order-processors",
};

import { SchemaRegistry, SchemaType } from '@confluent/kafka-javascript';

const schemaRegistry = new SchemaRegistry({
  host: 'http://schema-registry:8081',
  auth: {
    username: process.env.SR_USER,
    password: process.env.SR_PASSWORD,
  },
});

// Define an order event schema
const orderSchema = {
  type: 'record',
  name: 'Order',
  namespace: 'com.example.orders',
  fields: [
    { name: 'orderId', type: 'string' },
    { name: 'customerId', type: 'string' },
    { name: 'amount', type: 'double' },
    { name: 'timestamp', type: 'long' },
    // New field with default (backward compatible)
    { name: 'currency', type: 'string', default: 'USD' },
  ],
};

// Register schema
const schemaId = await schemaRegistry.register(
  SchemaType.AVRO,
  {
    type: 'AVRO',
    schema: JSON.stringify(orderSchema),
  },
  'orders-value'
);

// Serialize with schema
async function serializeOrder(order: any): Promise<Buffer> {
  const schemaId = await schemaRegistry.getSchemaId(
    'orders-value',
    SchemaType.AVRO
  );
  const serializer = schemaRegistry.getSerializer({ schemaId });
  return serializer.serialize(order);
}

// Deserialize with schema
async function deserializeOrder(buffer: Buffer): Promise<any> {
  const deserializer = schemaRegistry.getDeserializer();
  return deserializer.deserialize(buffer);
}

// Send Avro-serialized event
const serialized = await serializeOrder({
  orderId: 'order-123',
  customerId: 'cust-456',
  amount: 99.99,
  timestamp: Date.now(),
  currency: 'USD',
});

await producer.send({
  topic: 'orders',
  messages: [{ key: 'order-123', value: serialized }],
});

// Maven: io.confluent:kafka-avro-serializer
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;

@Service
public class SchemaRegistryProducerService {
    private final KafkaTemplate<String, GenericRecord> avroTemplate;

    private static final String ORDER_SCHEMA_JSON = """
        {
          "type": "record",
          "name": "Order",
          "namespace": "com.example.orders",
          "fields": [
            {"name": "orderId", "type": "string"},
            {"name": "customerId", "type": "string"},
            {"name": "amount", "type": "double"},
            {"name": "timestamp", "type": "long"},
            {"name": "currency", "type": "string", "default": "USD"}
          ]
        }""";

    private final Schema orderSchema = new Schema.Parser().parse(ORDER_SCHEMA_JSON);

    public void sendOrderEvent(String orderId, String customerId, double amount) {
        GenericRecord record = new GenericData.Record(orderSchema);
        record.put("orderId", orderId);
        record.put("customerId", customerId);
        record.put("amount", amount);
        record.put("timestamp", System.currentTimeMillis());
        record.put("currency", "USD");

        avroTemplate.send("orders", orderId, record);
    }
}

// Spring Boot config for Avro serializer
@Bean
public ProducerFactory<String, GenericRecord> avroProducerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
    config.put("schema.registry.url", "http://schema-registry:8081");
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
    return new DefaultKafkaProducerFactory<>(config);
}

# pip install confluent-kafka[avro] fastavro
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer, AvroDeserializer
from confluent_kafka import SerializingProducer, DeserializingConsumer
import os

schema_registry_client = SchemaRegistryClient({
    "url": "http://schema-registry:8081",
    "basic.auth.user.info": f"{os.getenv('SR_USER')}:{os.getenv('SR_PASSWORD')}",
})

order_schema_str = """
{
  "type": "record",
  "name": "Order",
  "namespace": "com.example.orders",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "timestamp", "type": "long"},
    {"name": "currency", "type": "string", "default": "USD"}
  ]
}"""

avro_serializer = AvroSerializer(schema_registry_client, order_schema_str)
avro_deserializer = AvroDeserializer(schema_registry_client)

producer = SerializingProducer({
    "bootstrap.servers": "kafka-1:9092",
    "key.serializer": lambda key, ctx: key.encode(),
    "value.serializer": avro_serializer,
})

# Send Avro-serialized event
producer.produce(
    topic="orders",
    key="order-123",
    value={
        "orderId": "order-123",
        "customerId": "cust-456",
        "amount": 99.99,
        "timestamp": int(time.time() * 1000),
        "currency": "USD",
    },
)
producer.flush()

// NuGet: Confluent.SchemaRegistry.Serdes.Avro
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;
using Confluent.Kafka;
using Avro;
using Avro.Generic;

var schemaRegistryConfig = new SchemaRegistryConfig
{
    Url = "http://schema-registry:8081",
    BasicAuthUserInfo = $"{Environment.GetEnvironmentVariable("SR_USER")}:{Environment.GetEnvironmentVariable("SR_PASSWORD")}",
};

using var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);

const string orderSchemaJson = """
{
  "type": "record",
  "name": "Order",
  "namespace": "com.example.orders",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "timestamp", "type": "long"},
    {"name": "currency", "type": "string", "default": "USD"}
  ]
}""";

var schema = (RecordSchema)Schema.Parse(orderSchemaJson);

var producerConfig = new ProducerConfig { BootstrapServers = "kafka-1:9092" };
using var producer = new ProducerBuilder<string, GenericRecord>(producerConfig)
    .SetValueSerializer(new AvroSerializer<GenericRecord>(schemaRegistry))
    .Build();

// Send Avro-serialized event
var record = new GenericRecord(schema);
record.Add("orderId", "order-123");
record.Add("customerId", "cust-456");
record.Add("amount", 99.99);
record.Add("timestamp", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds());
record.Add("currency", "USD");

await producer.ProduceAsync("orders", new Message<string, GenericRecord>
{
    Key = "order-123",
    Value = record,
});

// Event Sourcing: Store all events, derive state
interface OrderEvent {
  eventType: 'OrderCreated' | 'OrderPaid' | 'OrderShipped' | 'OrderCancelled';
  aggregateId: string;
  timestamp: number;
  data: any;
}

class OrderAggregate {
  private events: OrderEvent[] = [];
  private orderId: string;

  async applyEvent(event: OrderEvent): Promise<void> {
    this.events.push(event);

    // Send to event store (Kafka)
    await producer.sendEvent('order-events', event.aggregateId, event);

    // Update read models for CQRS
    await this.updateReadModel(event);
  }

  private async updateReadModel(event: OrderEvent): Promise<void> {
    switch (event.eventType) {
      case 'OrderCreated':
        await database.execute(
          'INSERT INTO order_summary (order_id, status) VALUES (?, ?)',
          [event.aggregateId, 'pending']
        );
        break;
      case 'OrderPaid':
        await database.execute(
          'UPDATE order_summary SET status = ? WHERE order_id = ?',
          ['paid', event.aggregateId]
        );
        break;
      case 'OrderShipped':
        await database.execute(
          'UPDATE order_summary SET status = ? WHERE order_id = ?',
          ['shipped', event.aggregateId]
        );
        break;
    }
  }

  // Rebuild state from event stream
  static async loadFromEvents(
    orderId: string,
    events: OrderEvent[]
  ): Promise<OrderAggregate> {
    const aggregate = new OrderAggregate();
    aggregate.orderId = orderId;
    aggregate.events = events;
    return aggregate;
  }
}

// Usage
const order = new OrderAggregate();
await order.applyEvent({
  eventType: 'OrderCreated',
  aggregateId: 'order-123',
  timestamp: Date.now(),
  data: {
    customerId: 'cust-456',
    items: [{ productId: 'prod-1', quantity: 2 }],
  },
});

// Event Sourcing: Store all events, derive state
public sealed interface OrderEvent permits OrderCreated, OrderPaid, OrderShipped, OrderCancelled {
    String aggregateId();
    long timestamp();
}

public record OrderCreated(String aggregateId, long timestamp, String customerId, List<OrderItem> items) implements OrderEvent {}
public record OrderPaid(String aggregateId, long timestamp, double amount) implements OrderEvent {}
public record OrderShipped(String aggregateId, long timestamp, String trackingNumber) implements OrderEvent {}

@Service
public class OrderAggregate {
    private final List<OrderEvent> events = new ArrayList<>();
    private final KafkaProducerService producerService;
    private final JdbcTemplate jdbcTemplate;

    public void applyEvent(OrderEvent event) throws Exception {
        events.add(event);
        // Send to event store (Kafka)
        producerService.sendEvent("order-events", event.aggregateId(),
            objectMapper.writeValueAsString(event)).get();
        // Update read models for CQRS
        updateReadModel(event);
    }

    private void updateReadModel(OrderEvent event) {
        switch (event) {
            case OrderCreated e -> jdbcTemplate.update(
                "INSERT INTO order_summary (order_id, status) VALUES (?, ?)",
                e.aggregateId(), "pending");
            case OrderPaid e -> jdbcTemplate.update(
                "UPDATE order_summary SET status = ? WHERE order_id = ?",
                "paid", e.aggregateId());
            case OrderShipped e -> jdbcTemplate.update(
                "UPDATE order_summary SET status = ? WHERE order_id = ?",
                "shipped", e.aggregateId());
            default -> {}
        }
    }

    public static OrderAggregate loadFromEvents(String orderId, List<OrderEvent> events) {
        var aggregate = new OrderAggregate(null, null);
        aggregate.events.addAll(events);
        return aggregate;
    }
}