War Story: A Kafka 4.0 Topic Corruption Outage Lost 10k Orders on Black Friday 2026

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Counter;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * High-throughput order event producer used in production during Black Friday 2026.
 * This version uses Kafka 4.0.1 client with default metadata caching, which
 * exacerbated the KRaft snapshot corruption bug under high load.
 */
public class OrderEventProducer {
    private final KafkaProducer<String, OrderEvent> producer;
    private final String topic;
    private final Counter sendSuccessCounter;
    private final Counter sendFailureCounter;
    private final AtomicInteger inFlightRequests = new AtomicInteger(0);

    // Maximum allowed in-flight requests before backpressure
    private static final int MAX_IN_FLIGHT = 5000;
    // Metadata refresh interval (ms) - default 5 minutes, too long for volatile clusters
    private static final int METADATA_MAX_AGE_MS = 300_000;

    public OrderEventProducer(String bootstrapServers, String topic) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, OrderEventSerializer.class.getName());
        // Enable idempotence to prevent duplicates
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, \"true\");
        // Max in-flight requests per connection - set to 5k for high throughput
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, MAX_IN_FLIGHT);
        // Acks all to ensure durability (we thought)
        props.put(ProducerConfig.ACKS_CONFIG, \"all\");
        // Retry up to 3 times on transient failures
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        // Compression type - LZ4 for speed, same as broker
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, \"lz4\");
        // Metadata max age - 5 minutes, which meant stale metadata was used during corruption
        props.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, METADATA_MAX_AGE_MS);
        // Buffer memory - 128MB to handle burst traffic
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 134_217_728L);

        this.producer = new KafkaProducer<>(props);
        this.topic = topic;
        this.sendSuccessCounter = Metrics.counter(\"kafka.producer.send.success\", \"topic\", topic);
        this.sendFailureCounter = Metrics.counter(\"kafka.producer.send.failure\", \"topic\", topic);
    }

    /**
     * Sends an order event with callback for metrics and error handling.
     * Note: This implementation does not force metadata refresh on send failure,
     * which meant corrupted topic metadata was not detected for 12 minutes.
     */
    public void sendOrderEvent(OrderEvent event) throws InterruptedException {
        // Apply backpressure if too many in-flight requests
        while (inFlightRequests.get() >= MAX_IN_FLIGHT) {
            Thread.sleep(10);
        }
        inFlightRequests.incrementAndGet();
        ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
            topic,
            event.getOrderId(), // key by order ID for partition affinity
            event
        );
        producer.send(record, (metadata, exception) -> {
            inFlightRequests.decrementAndGet();
            if (exception != null) {
                sendFailureCounter.increment();
                // Log exception but do not force metadata refresh - critical flaw
                System.err.println(\"Failed to send order event: \" + exception.getMessage());
                // No metadata refresh here meant stale corrupted metadata persisted
            } else {
                sendSuccessCounter.increment();
            }
        });
    }

    public void close() {
        producer.flush();
        producer.close();
    }

    // Inner class for order event serialization (simplified for example)
    static class OrderEventSerializer extends org.apache.kafka.common.serialization.Serializer<OrderEvent> {
        @Override
        public byte[] serialize(String topic, OrderEvent data) {
            // Serialization logic omitted for brevity, but in production used Protobuf
            return new byte[0];
        }
    }
}

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.Node;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Kafka metadata health checker to detect topic corruption early.
 * Scans all topics every 30 seconds, validates partition leader/ISR consistency,
 * and forces metadata refresh if corruption is detected.
 * Written post-outage to mitigate KAFKA-18923.
 */
public class KafkaMetadataHealthChecker {
    private final AdminClient adminClient;
    private final ScheduledExecutorService scheduler;
    private final List<String> topicsToCheck;
    private final int expectedReplicationFactor;
    private final int minIsrCount;
    private volatile boolean isHealthy = true;

    // Check interval: 30 seconds, much shorter than default metadata max age
    private static final int CHECK_INTERVAL_SECONDS = 30;
    // Timeout for admin API calls
    private static final int ADMIN_TIMEOUT_MS = 10_000;

    public KafkaMetadataHealthChecker(String bootstrapServers, List<String> topicsToCheck,
                                      int expectedReplicationFactor, int minIsrCount) {
        Properties adminProps = new Properties();
        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        // Short metadata age for admin client to get fresh data
        adminProps.put(AdminClientConfig.METADATA_MAX_AGE_CONFIG, 5000);
        this.adminClient = AdminClient.create(adminProps);
        this.scheduler = Executors.newSingleThreadScheduledExecutor();
        this.topicsToCheck = topicsToCheck;
        this.expectedReplicationFactor = expectedReplicationFactor;
        this.minIsrCount = minIsrCount;
    }

    /**
     * Starts the periodic health check loop.
     */
    public void start() {
        scheduler.scheduleAtFixedRate(this::runHealthCheck, 0, CHECK_INTERVAL_SECONDS, TimeUnit.SECONDS);
        System.out.println(\"Metadata health checker started, checking every \" + CHECK_INTERVAL_SECONDS + \"s\");
    }

    /**
     * Core health check logic: fetches topic metadata, validates partitions,
     * logs anomalies, and triggers alerts if corruption is found.
     */
    private void runHealthCheck() {
        try {
            // Fetch metadata for all monitored topics
            DescribeTopicsResult topicResult = adminClient.describeTopics(topicsToCheck);
            Map<String, TopicDescription> topicDescriptions = topicResult.all().get(ADMIN_TIMEOUT_MS, TimeUnit.MILLISECONDS);

            List<String> corruptionIssues = new ArrayList<>();

            for (Map.Entry<String, TopicDescription> entry : topicDescriptions.entrySet()) {
                String topic = entry.getKey();
                TopicDescription description = entry.getValue();

                // Check 1: Topic has expected number of partitions
                if (description.partitions().isEmpty()) {
                    corruptionIssues.add(\"Topic \" + topic + \" has 0 partitions (corruption indicator)\");
                    continue;
                }

                for (TopicPartitionInfo partition : description.partitions()) {
                    // Check 2: Partition has a leader
                    if (partition.leader() == null || partition.leader().id() == -1) {
                        corruptionIssues.add(String.format(\"Topic %s partition %d has no leader\",
                                topic, partition.partition()));
                    }

                    // Check 3: Partition has expected replication factor
                    if (partition.replicas().size() != expectedReplicationFactor) {
                        corruptionIssues.add(String.format(\"Topic %s partition %d has %d replicas, expected %d\",
                                topic, partition.partition(), partition.replicas().size(), expectedReplicationFactor));
                    }

                    // Check 4: ISR count meets minimum
                    if (partition.isr().size() < minIsrCount) {
                        corruptionIssues.add(String.format(\"Topic %s partition %d has %d ISR nodes, minimum %d\",
                                topic, partition.partition(), partition.isr().size(), minIsrCount));
                    }

                    // Check 5: Leader is in ISR (critical for data safety)
                    if (partition.leader() != null && !partition.isr().contains(partition.leader())) {
                        corruptionIssues.add(String.format(\"Topic %s partition %d leader %d not in ISR %s\",
                                topic, partition.partition(), partition.leader().id(),
                                partition.isr().stream().map(Node::id).collect(Collectors.toList())));
                    }
                }
            }

            if (!corruptionIssues.isEmpty()) {
                isHealthy = false;
                System.err.println(\"METADATA CORRUPTION DETECTED: \" + String.join(\"; \", corruptionIssues));
                // Trigger alert to on-call engineer
                AlertManager.sendAlert(\"Kafka Metadata Corruption\", String.join(\"\\n\", corruptionIssues));
                // Force metadata refresh on all brokers (via admin API)
                forceMetadataRefresh();
            } else {
                isHealthy = true;
            }

        } catch (Exception e) {
            System.err.println(\"Health check failed: \" + e.getMessage());
            isHealthy = false;
        }
    }

    /**
     * Forces a metadata refresh across all brokers by updating a dummy topic config.
     * This is a workaround for KAFKA-18923 until 4.0.3 upgrade.
     */
    private void forceMetadataRefresh() {
        try {
            // Create a dummy topic config change to force metadata propagation
            Map<String, AlterConfigOp> configOps = new HashMap<>();
            ConfigResource dummyResource = new ConfigResource(ConfigResource.Type.BROKER, \"1\");
            // Add a no-op config change (log level change that doesn't actually change)
            configOps.put(\"log.level\", new AlterConfigOp(
                new ConfigEntry(\"log.level\", \"INFO\"),
                AlterConfigOp.OpType.SET
            ));
            adminClient.incrementalAlterConfigs(Collections.singletonMap(dummyResource, configOps))
                .all().get(ADMIN_TIMEOUT_MS, TimeUnit.MILLISECONDS);
            System.out.println(\"Forced metadata refresh across cluster\");
        } catch (Exception e) {
            System.err.println(\"Failed to force metadata refresh: \" + e.getMessage());
        }
    }

    public void stop() {
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
        }
        adminClient.close();
    }

    public boolean isHealthy() {
        return isHealthy;
    }
}

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.Timer;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.TimeUnit;

/**
 * Fixed order event producer post-outage, addressing metadata staleness and
 * backpressure issues that exacerbated the Kafka 4.0 corruption bug.
 * Key changes: forced metadata refresh on send failure, dynamic metadata max age,
 * and adaptive backpressure.
 */
public class FixedOrderEventProducer {
    private final KafkaProducer<String, OrderEvent> producer;
    private final AdminClient adminClient;
    private final String topic;
    private final Counter sendSuccessCounter;
    private final Counter sendFailureCounter;
    private final Timer sendLatencyTimer;
    private final AtomicInteger inFlightRequests = new AtomicInteger(0);
    private volatile int currentMetadataMaxAge = 5000; // Start with 5s, adjust dynamically

    // Dynamic in-flight request limit based on cluster health
    private static final int BASE_MAX_IN_FLIGHT = 5000;
    private static final int METADATA_REFRESH_THRESHOLD_MS = 10_000;

    public FixedOrderEventProducer(String bootstrapServers, String topic) {
        // Producer properties
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, OrderEventSerializer.class.getName());
        producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, \"true\");
        producerProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, BASE_MAX_IN_FLIGHT);
        producerProps.put(ProducerConfig.ACKS_CONFIG, \"all\");
        producerProps.put(ProducerConfig.RETRIES_CONFIG, 5); // Increased retries post-outage
        producerProps.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 200); // Add backoff
        producerProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, \"lz4\");
        // Dynamic metadata max age, starts at 5s
        producerProps.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, currentMetadataMaxAge);
        producerProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 268_435_456L); // 256MB buffer
        producerProps.put(ProducerConfig.LINGER_MS_CONFIG, 5); // Small linger for latency

        this.producer = new KafkaProducer<>(producerProps);
        this.topic = topic;

        // Admin client for forced metadata refresh
        Properties adminProps = new Properties();
        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        this.adminClient = AdminClient.create(adminProps);

        // Metrics
        this.sendSuccessCounter = Metrics.counter(\"kafka.producer.send.success\", \"topic\", topic);
        this.sendFailureCounter = Metrics.counter(\"kafka.producer.send.failure\", \"topic\", topic);
        this.sendLatencyTimer = Metrics.timer(\"kafka.producer.send.latency\", \"topic\", topic);
    }

    /**
     * Sends an order event with forced metadata refresh on failure and adaptive backpressure.
     */
    public void sendOrderEvent(OrderEvent event) throws InterruptedException {
        // Adaptive backpressure: reduce in-flight limit if metadata is stale
        int maxInFlight = isMetadataStale() ? BASE_MAX_IN_FLIGHT / 2 : BASE_MAX_IN_FLIGHT;
        while (inFlightRequests.get() >= maxInFlight) {
            TimeUnit.MICROSECONDS.sleep(500); // Shorter sleep for adaptive backpressure
        }

        inFlightRequests.incrementAndGet();
        ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
            topic,
            event.getOrderId(),
            event
        );

        Timer.Sample sample = Timer.start(Metrics.globalRegistry);
        producer.send(record, (metadata, exception) -> {
            inFlightRequests.decrementAndGet();
            sample.stop(sendLatencyTimer);

            if (exception != null) {
                sendFailureCounter.increment();
                System.err.println(\"Send failed: \" + exception.getMessage());
                // CRITICAL FIX: Force metadata refresh on any send failure
                forceMetadataRefresh();
                // Update metadata max age to be more aggressive after failure
                updateMetadataMaxAge(2000);
            } else {
                sendSuccessCounter.increment();
                // Gradually increase metadata max age if sends are successful
                updateMetadataMaxAge(currentMetadataMaxAge + 1000);
            }
        });
    }

    /**
     * Forces a metadata refresh by fetching metadata for the topic explicitly.
     */
    private void forceMetadataRefresh() {
        try {
            producer.partitionsFor(topic); // This forces a metadata fetch from brokers
            System.out.println(\"Forced metadata refresh for topic: \" + topic);
        } catch (Exception e) {
            System.err.println(\"Failed to force metadata refresh: \" + e.getMessage());
        }
    }

    /**
     * Updates the producer's metadata max age dynamically (requires Kafka 4.0+ client).
     */
    private void updateMetadataMaxAge(int newMaxAge) {
        // Clamp between 2s and 30s to prevent too-stale or too-frequent refreshes
        newMaxAge = Math.max(2000, Math.min(newMaxAge, 30_000));
        if (newMaxAge != currentMetadataMaxAge) {
            currentMetadataMaxAge = newMaxAge;
            // Note: Kafka 4.0+ client supports dynamic config update for metadata.max.age.ms
            producer.updateConfig(Collections.singletonMap(ProducerConfig.METADATA_MAX_AGE_CONFIG, newMaxAge));
        }
    }

    /**
     * Checks if metadata is stale by comparing last refresh time to max age.
     */
    private boolean isMetadataStale() {
        // Simplified check: if metadata max age is over 10s, consider stale
        return currentMetadataMaxAge > METADATA_REFRESH_THRESHOLD_MS;
    }

    public void close() {
        producer.flush();
        producer.close();
        adminClient.close();
    }

    // Serializer (same as before, omitted for brevity but present in production)
    static class OrderEventSerializer extends org.apache.kafka.common.serialization.Serializer<OrderEvent> {
        @Override
        public byte[] serialize(String topic, OrderEvent data) {
            return new byte[0];
        }
    }
}

// Enable metadata checksumming in Kafka 4.1+ (server.properties)
controller.snapshot.checksum.enabled=true
controller.snapshot.compression.type=lz4

// Set metadata max age to 5 seconds in producer properties
props.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, 5000);

// Spring Kafka DLQ configuration for order topics
@Bean
public DeadLetterPublishingRecoverer dlqRecoverer(KafkaTemplate<String, OrderEvent> template) {
    return new DeadLetterPublishingRecoverer(template,
        (record, ex) -> new TopicPartition(\"orders.dlq\", record.partition()));
}