// Emergency config change (zero-downtime via Spring Cloud Config)
management.kafka.consumer.max-poll-records=10  // Reduced from 500 to 10
// This reduces per-poll processing time from 1000s to 20s
// Rebalancing stops immediately after config propagates

// Fix 1: Decouple poll frequency from processing time
// Use separate thread for processing - consumer thread only polls
 
@Configuration
public class ResilientConsumerConfig {
 
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, PaymentEvent> factory(
            ConsumerFactory<String, PaymentEvent> consumerFactory) {
 
        ConcurrentKafkaListenerContainerFactory<String, PaymentEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
 
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3);
 
        // CRITICAL: Set max.poll.records small
        // Process fewer messages per poll to stay within max.poll.interval.ms
        // Rule: max.poll.records * worst_case_processing_time_ms < max.poll.interval.ms
        factory.getContainerProperties().setConsumerProperties(
            propertiesBuilder()
                .put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50)           // Reduced
                .put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000)   // 5 minutes
                .put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000)      // 45 seconds
                .put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15000)   // 15 seconds
                .build()
        );
 
        return factory;
    }
}
 
// Fix 2: Async processing with bounded queue
@KafkaListener(topics = "payment-commands", containerFactory = "resilientFactory")
public void handlePayment(PaymentEvent event, Acknowledgment ack) {
    // Submit to bounded executor - never block the poll thread
    CompletableFuture.runAsync(() -> {
        try {
            paymentProcessor.processWithTimeout(event, Duration.ofSeconds(10));
            ack.acknowledge();
        } catch (Exception e) {
            log.error("Payment processing failed: {}", event.getPaymentId(), e);
            // Do NOT ack - message redelivered after visibility timeout
        }
    }, paymentExecutor);  // paymentExecutor has bounded queue and rejection policy
}

// Producer code (the culprit)
public void publishOrder(Order order) {
    String partitionKey = order.getMerchantId();  // Used as partition key
    kafkaTemplate.send("order-events", partitionKey, event);
}
 
// Problem: Top 3 merchants by order volume:
// Merchant ID "AMAZON-MARKETPLACE": 40% of all orders
// Merchant ID "WALMART-MARKETPLACE": 25% of all orders
// Merchant ID "EBAY-MARKETPLACE": 15% of all orders
 
// hash("AMAZON-MARKETPLACE") % 12 = 2  (maps to Partition 2!)
// 40% of all orders → single partition → single consumer → bottleneck
 
// Adding more consumers does nothing:
// Extra consumers for partition 2 = impossible (one consumer per partition in Kafka)
// Other 11 partitions are idle

# Detect hot partitions
kafka-consumer-groups.sh --describe --group order-processor \
  | awk '{print $4, $5}' | sort -k2 -rn | head -5
# Shows partitions sorted by lag - hot partition will stand out
 
# Per-partition throughput via JMX/Prometheus
kafka_partition_records_count{topic="order-events"}
# One partition showing 10x the count of others = hot partition

// Fix 1: Compound partition key (merchant + random suffix)
// Groups merchant orders together for ordering, but distributes across multiple partitions
 
public void publishOrder(Order order) {
    // Instead of just merchantId, use merchantId + bucket number
    // Each merchant gets up to 4 "virtual lanes"
    int bucket = order.getOrderId().hashCode() & 3;  // 0, 1, 2, or 3
    String partitionKey = order.getMerchantId() + "-" + bucket;
 
    kafkaTemplate.send("order-events", partitionKey, event);
}
 
// "AMAZON-MARKETPLACE" now maps to 4 different keys:
// "AMAZON-MARKETPLACE-0" → any partition
// "AMAZON-MARKETPLACE-1" → any partition
// "AMAZON-MARKETPLACE-2" → any partition
// "AMAZON-MARKETPLACE-3" → any partition
// 40% of orders now spread across 4 partitions instead of 1
 
// Trade-off: within-merchant ordering is NOT guaranteed across buckets
// If order-level ordering matters within a merchant: use orderId as partition key instead

// Fix 2: Custom Partitioner for known hot keys
@Component
public class LoadBalancedPartitioner implements Partitioner {
 
    private static final Map<String, Integer> PARTITION_COUNT_FOR_HOT_MERCHANTS = Map.of(
        "AMAZON-MARKETPLACE", 4,    // Give Amazon 4 partition slots
        "WALMART-MARKETPLACE", 2,   // Give Walmart 2 partition slots
        "EBAY-MARKETPLACE", 2       // Give eBay 2 partition slots
    );
 
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        String merchantId = extractMerchantId(key.toString());
        int totalPartitions = cluster.partitionCountForTopic(topic);
 
        if (PARTITION_COUNT_FOR_HOT_MERCHANTS.containsKey(merchantId)) {
            int slots = PARTITION_COUNT_FOR_HOT_MERCHANTS.get(merchantId);
            // Distribute hot merchant across multiple partitions
            int basePartition = Math.abs(merchantId.hashCode()) % totalPartitions;
            int slot = Math.abs(key.toString().hashCode()) % slots;
            return (basePartition + slot) % totalPartitions;
        }
 
        // Default: standard key-based hashing for all other merchants
        return Math.abs(key.toString().hashCode()) % totalPartitions;
    }
}

# If topic must be repartitioned (more partitions for better distribution):
# Step 1: Create new topic with more partitions
kafka-topics.sh --create --topic order-events-v2 --partitions 48
 
# Step 2: Run dual-write (producer publishes to both topics) for 24 hours
# Step 3: Start new consumer groups on order-events-v2
# Step 4: Validate new consumers are caught up and healthy
# Step 5: Stop old consumer groups (they're caught up to 0 lag)
# Step 6: Remove dual-write from producer
# Step 7: Delete old topic after retention period

@Service
@Slf4j
public class IdempotentPaymentConsumer {
 
    private final PaymentIdempotencyRepository idempotencyRepo;
    private final StripePaymentGateway stripe;
    private final RedissonClient redisson;
 
    @KafkaListener(topics = "payment-commands", groupId = "payment-processor")
    @Transactional
    public void processPayment(
            PaymentCommandEvent event,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {
 
        String idempotencyKey = event.getPaymentId();
        String lockKey = "payment-lock:" + idempotencyKey;
 
        // Distributed lock: only ONE instance can process this paymentId at a time
        // Prevents concurrent consumer-7 + consumer-8 race condition
        RLock lock = redisson.getLock(lockKey);
 
        try {
            boolean locked = lock.tryLock(5, 30, TimeUnit.SECONDS);
            if (!locked) {
                log.warn("Could not acquire lock for paymentId={}, will retry", idempotencyKey);
                // Do NOT ack - Kafka will redeliver
                return;
            }
 
            // Check if already processed (handles redelivery after offset commit failure)
            Optional<ProcessedPayment> existing = idempotencyRepo.findById(idempotencyKey);
            if (existing.isPresent()) {
                log.info("Duplicate payment command ignored: paymentId={}, " +
                         "originally processed at partition={}, offset={}",
                         idempotencyKey,
                         existing.get().getOriginalPartition(),
                         existing.get().getOriginalOffset());
                ack.acknowledge(); // Ack so we don't keep reprocessing
                return;
            }
 
            // Process (with idempotency key passed to Stripe)
            ChargeResult result = stripe.charge(
                StripeChargeRequest.builder()
                    .amount(event.getAmount())
                    .customerId(event.getCustomerId())
                    .idempotencyKey(idempotencyKey)  // Stripe-level idempotency
                    .build()
            );
 
            // Record as processed BEFORE acking
            idempotencyRepo.save(ProcessedPayment.builder()
                .paymentId(idempotencyKey)
                .stripeChargeId(result.getChargeId())
                .originalPartition(partition)
                .originalOffset(offset)
                .processedAt(Instant.now())
                .build());
 
            ack.acknowledge();
            log.info("Payment processed: paymentId={}, chargeId={}", idempotencyKey, result.getChargeId());
 
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("Interrupted while waiting for lock: paymentId={}", idempotencyKey);
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

# Step 1: Calculate actual consumer throughput
kafka-consumer-groups.sh --describe --group analytics-pipeline | \
  awk 'NR>1 {print $1, $2, $6}' | \
  awk '{lag[$2]+=$6} END {for(t in lag) print t, lag[t]}'
# Shows total lag per topic
 
# Step 2: Consumer processing time (application metrics)
# histogram_quantile(0.99, kafka_consumer_fetch_latency_seconds)
# This shows HOW LONG consumers spend processing, not just polling
 
# Step 3: Throughput trend
# rate(kafka_consumer_records_consumed_total[5m])
# If this is flat or declining as you add consumers: contention somewhere
 
# Step 4: Identify the bottleneck - thread dump of consumer
jstack <consumer-pid> | grep -A 20 "WAITING\|TIMED_WAITING"
# Shows threads blocked on Redis connection pool

// Fix 1: Batch Redis calls (one pipeline call for 100 messages instead of 100 individual calls)
@KafkaListener(
    topics = "clickstream-events",
    containerFactory = "batchConsumerFactory"  // Batch listener mode
)
public void handleBatch(List<ConsumerRecord<String, ClickEvent>> records, Acknowledgment ack) {
    // Collect all user IDs from the batch
    Set<String> userIds = records.stream()
        .map(r -> r.value().getUserId())
        .collect(toSet());
 
    // One batched Redis call for all users in the batch
    // Replaces N individual calls with 1 pipeline call
    Map<String, UserMetadata> userMetadata = redisClient.pipeline(pipe -> {
        userIds.forEach(id -> pipe.get("user:" + id));
    });
 
    // Process records with pre-fetched metadata
    List<EnrichedClickEvent> enriched = records.stream()
        .map(r -> enrich(r.value(), userMetadata.get(r.value().getUserId())))
        .collect(toList());
 
    clickhouseWriter.writeBatch(enriched);
    ack.acknowledge();
}
 
// Fix 2: Right-size connection pool for the actual concurrency
@Bean
public RedisConnectionFactory redisConnectionFactory() {
    LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()
        .clientResources(DefaultClientResources.builder()
            .ioThreadPoolSize(8)
            .computationThreadPoolSize(8)
            .build())
        .build();
 
    // Pool size = number of consumer threads × 1.5 (to avoid starvation)
    LettucePoolingClientConfiguration poolingConfig = LettucePoolingClientConfiguration.builder()
        .poolConfig(new GenericObjectPoolConfig<>()
            .maxTotal(40)     // 24 consumer threads × 1.5 = 36, rounded up
            .maxIdle(24)
            .minIdle(10)
            .maxWaitMillis(2000))
        .build();
 
    return new LettuceConnectionFactory(redisStandaloneConfig, poolingConfig);
}
 
// Result: 180ms Redis call → 5ms average (batch eliminated per-message call overhead)
// Throughput: 295 msg/min → 18,000 msg/min per consumer (60x improvement)
// Lag recovery: 8 billion messages recovered in 18 hours instead of never

// Fix 1: Schema Registry cluster (multi-instance, multi-AZ)
# docker-compose / Kubernetes - NOT a single instance
 
schema-registry-1:
  image: confluentinc/cp-schema-registry:7.4.0
  environment:
    SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092
    SCHEMA_REGISTRY_HOST_NAME: schema-registry-1
  deploy:
    replicas: 3                    # 3 instances
    placement:
      constraints:
        - node.labels.az != same   # Each in different AZ
 
# Load balancer in front of schema-registry instances
# A single registry instance failing does not cause service interruption
 
// Fix 2: Producer-side graceful degradation
@Configuration
public class ResilientAvroProducerConfig {
 
    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBrokers);
 
        // Schema Registry with timeout and fallback
        config.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
            "http://schema-registry-1:8081,http://schema-registry-2:8081,http://schema-registry-3:8081");
 
        // Aggressive local caching - reduces Registry calls to near-zero at steady state
        config.put(AbstractKafkaSchemaSerDeConfig.MAX_SCHEMAS_PER_SUBJECT_CONFIG, 100);
 
        // Use latest cached schema if Registry temporarily unavailable
        config.put("schema.registry.request.timeout.ms", "3000");
 
        // Pre-warm cache at startup (not lazily)
        config.put(AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS, false);
 
        return new DefaultKafkaProducerFactory<>(config);
    }
}
 
// Fix 3: Startup schema pre-loading
@Component
public class SchemaRegistryCacheWarmer implements ApplicationRunner {
 
    private final SchemaRegistryClient schemaRegistryClient;
    private final List<String> criticalSchemaSubjects;
 
    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("Pre-loading schemas from registry...");
        for (String subject : criticalSchemaSubjects) {
            try {
                // Force-populate local cache
                schemaRegistryClient.getLatestSchemaMetadata(subject);
                log.info("Loaded schema: {}", subject);
            } catch (Exception e) {
                // Log but do not fail startup - use locally bundled fallback schema
                log.warn("Could not load schema {} from registry, using bundled fallback", subject);
                schemaRegistryClient.register(subject, bundledSchema(subject));
            }
        }
    }
}

// Fix 1: Ensure MirrorMaker 2 preserves source ordering
# MirrorMaker 2 configuration
clusters = source, target
source.bootstrap.servers = us-east-kafka:9092
target.bootstrap.servers = eu-west-kafka:9092
 
# CRITICAL: Use same number of replication tasks as source partitions
# One task per partition ensures in-order replication within partitions
source->target.tasks.max = 12  # Equal to partition count
 
# Fix 2: EU-West consumers use source-timestamp ordering, not arrival order
@Service
public class FinancialEventConsumer {
 
    private final FinancialEventBuffer eventBuffer;
 
    @KafkaListener(topics = "account-events.mirror", groupId = "eu-balance-processor")
    public void handle(AccountEvent event,
                       @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long kafkaTimestamp) {
 
        // Use source timestamp (from US-East), NOT EU Kafka arrival timestamp
        long sourceTimestamp = event.getSourceTimestamp(); // Embedded in event payload
 
        // Buffer events with out-of-order tolerance window (500ms)
        eventBuffer.add(event, sourceTimestamp);
    }
}
 
@Component
public class FinancialEventBuffer {
 
    private final PriorityQueue<BufferedEvent> buffer =
        new PriorityQueue<>(Comparator.comparingLong(BufferedEvent::getSourceTimestamp));
 
    private final Duration OUT_OF_ORDER_TOLERANCE = Duration.ofMillis(500);
    private volatile Instant lastProcessed = Instant.now();
 
    public void add(AccountEvent event, long sourceTimestamp) {
        buffer.offer(new BufferedEvent(event, sourceTimestamp));
        drainSafeEvents();
    }
 
    private void drainSafeEvents() {
        long nowMs = Instant.now().toEpochMilli();
        while (!buffer.isEmpty()) {
            BufferedEvent head = buffer.peek();
            // Only process events that are "old enough" to be considered ordered
            // Any events timestamped within the last 500ms might still be followed by
            // earlier-timestamped events still in transit from US-East
            if (nowMs - head.getSourceTimestamp() > OUT_OF_ORDER_TOLERANCE.toMillis()) {
                buffer.poll();
                processEvent(head.getEvent());
            } else {
                break; // Head of queue is too recent - wait for more events
            }
        }
    }
}

# EMERGENCY: Kafka brokers rejecting all new messages
 
# Step 1: Identify which topics are consuming most disk
kafka-log-dirs.sh --bootstrap-server broker:9092 --describe \
  | grep -v '^{' | sort -k3 -rn | head -20
# Shows topics sorted by disk usage
 
# Step 2: Emergency retention reduction on the runaway topic
kafka-configs.sh --alter \
  --entity-type topics \
  --entity-name analytics-raw-events \
  --add-config retention.ms=3600000  # Drop to 1 hour IMMEDIATELY
# Kafka's log cleaner will start deleting old segments within minutes
 
# Step 3: Monitor disk recovery
watch -n 5 'df -h /var/kafka-logs'
# Should see disk usage declining as log cleaner works
 
# Step 4: Once disk below 80%, restore reasonable retention
kafka-configs.sh --alter \
  --entity-type topics \
  --entity-name analytics-raw-events \
  --add-config retention.ms=86400000  # 24 hours (appropriate for analytics)
  --add-config retention.bytes=52428800000  # 50GB cap regardless of time
 
# Step 5: Verify producers can send again (test message)
kafka-console-producer.sh --broker-list broker:9092 \
  --topic analytics-raw-events <<< "test"

// Enforce retention policy at topic creation time
@Component
public class TopicCreationEnforcer {
 
    private static final Map<String, RetentionPolicy> POLICIES = Map.of(
        "financial",   RetentionPolicy.of(Duration.ofDays(90), 100_000_000_000L),  // 90 days, 100GB
        "order",       RetentionPolicy.of(Duration.ofDays(30),  50_000_000_000L),  // 30 days, 50GB
        "analytics",   RetentionPolicy.of(Duration.ofDays(7),   20_000_000_000L),  // 7 days, 20GB
        "audit",       RetentionPolicy.of(Duration.ofDays(730), Long.MAX_VALUE),    // 2 years, unlimited
        "default",     RetentionPolicy.of(Duration.ofDays(3),   10_000_000_000L)   // 3 days, 10GB
    );
 
    @EventListener
    public void onTopicCreation(TopicCreationRequestedEvent event) {
        String topicCategory = inferCategory(event.getTopicName());
        RetentionPolicy policy = POLICIES.getOrDefault(topicCategory, POLICIES.get("default"));
 
        // BLOCK topic creation without retention policy
        if (!event.hasRetentionPolicy()) {
            throw new TopicPolicyViolationException(
                "Topic " + event.getTopicName() + " must specify retention. " +
                "Recommended policy for '" + topicCategory + "': " + policy
            );
        }
 
        // Warn if retention exceeds policy
        if (event.getRetentionMs() > policy.getMaxRetentionMs()) {
            alertingService.sendWarning(
                "Topic " + event.getTopicName() + " requests " +
                event.getRetentionDays() + " day retention. " +
                "Policy max for '" + topicCategory + "' is " + policy.getMaxDays() + " days."
            );
        }
    }
}

# Disk utilization alert - add to monitoring
groups:
  - name: kafka-disk-alerts
    rules:
      - alert: KafkaBrokerDiskUsageHigh
        expr: (node_filesystem_size_bytes{mountpoint="/var/kafka-logs"}
          - node_filesystem_avail_bytes{mountpoint="/var/kafka-logs"})
          / node_filesystem_size_bytes{mountpoint="/var/kafka-logs"} > 0.70
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Kafka broker disk {{ $labels.instance }} at {{ $value | humanizePercentage }}"
 
      - alert: KafkaBrokerDiskUsageCritical
        expr: ... > 0.85
        labels:
          severity: critical
        annotations:
          summary: "CRITICAL: Kafka broker disk nearly full - broker will stop accepting messages at 100%"

# NEVER use these settings for critical data:
# acks=1 (leader only - loses data on leader failure before replication)
# unclean.leader.election.enable=true (elects out-of-sync replicas, losing messages)
 
# ALWAYS use for critical data:
 
# Producer configuration
acks = all                    # Wait for ALL in-sync replicas to acknowledge
enable.idempotence = true      # Prevent duplicates from producer retries
max.in.flight.requests.per.connection = 1  # Strict ordering with idempotence
 
# Broker configuration (set globally and per-topic)
default.replication.factor = 3
min.insync.replicas = 2       # At least 2 replicas must acknowledge before success
unclean.leader.election.enable = false  # NEVER elect out-of-sync replica as leader
 
# Topic-level override for financial topics
kafka-configs.sh --alter \
  --entity-type topics \
  --entity-name order-events \
  --add-config min.insync.replicas=2 \
  --add-config unclean.leader.election.enable=false

// BROKEN: The actual production code that caused the incident
 
@KafkaListener(topics = "payment-events")
public void handlePayments(List<PaymentEvent> events) {
    // Process each event asynchronously for throughput
    List<CompletableFuture<Void>> futures = events.stream()
        .map(event -> CompletableFuture.runAsync(
            () -> paymentService.processPayment(event),
            processingExecutor
        ))
        .collect(toList());
 
    // WRONG: This does NOT wait for all futures to complete
    // The @KafkaListener method returns immediately
    // Spring Kafka sees the method returned without exception
    // If using auto-commit or BATCH ack mode: offsets committed NOW
    // The async tasks are still running in processingExecutor
    // If the application restarts in the next 500ms:
    //   - Offsets were committed (consumer thinks these are done)
    //   - But 30% of async tasks haven't completed yet
    //   - Those payments are LOST (offset committed, won't be redelivered)
}

// CORRECT: Batch listener with explicit acknowledgment after all async work completes
 
@KafkaListener(
    topics = "payment-events",
    containerFactory = "batchManualAckFactory"
)
public void handlePayments(
        List<ConsumerRecord<String, PaymentEvent>> records,
        Acknowledgment ack) {
 
    // Launch all async tasks
    List<CompletableFuture<PaymentResult>> futures = records.stream()
        .map(record -> CompletableFuture.supplyAsync(
            () -> paymentService.processPayment(record.value()),
            processingExecutor
        ))
        .collect(toList());
 
    // Wait for ALL futures to complete before acknowledging
    CompletableFuture<Void> allDone = CompletableFuture.allOf(
        futures.toArray(new CompletableFuture[0])
    );
 
    try {
        // CRITICAL: Block until ALL async processing is done
        // Only then: commit the offset
        allDone.get(60, TimeUnit.SECONDS);
        ack.acknowledge();  // Safe to commit now - all processing complete
 
    } catch (TimeoutException e) {
        log.error("Batch processing timed out - NOT committing offset, will retry");
        // Do NOT ack - Kafka redelivers the entire batch
        // Some might be processed twice - must be idempotent
 
    } catch (ExecutionException e) {
        log.error("Batch processing failed - checking individual results");
        // Check which futures succeeded and which failed
        handlePartialBatchFailure(records, futures, ack);
    }
}
 
private void handlePartialBatchFailure(
        List<ConsumerRecord<String, PaymentEvent>> records,
        List<CompletableFuture<PaymentResult>> futures,
        Acknowledgment ack) {
 
    // Find the first failed record
    for (int i = 0; i < futures.size(); i++) {
        if (futures.get(i).isCompletedExceptionally()) {
            // Commit offsets only up to the failed record
            // Records after the failed one will be redelivered
            if (i > 0) {
                ack.nack(i - 1, Duration.ofSeconds(1)); // Commit up to i-1
            }
            return;
        }
    }
}

// Consumer with startup rate limiting
 
@Component
public class ThrottledNotificationConsumer {
 
    private static final int NORMAL_RATE_PER_SECOND = 83;  // 5000/minute ÷ 60
    private static final int MAX_SENDGRID_RATE = 150;       // Never exceed this
 
    private final AtomicBoolean rampUpComplete = new AtomicBoolean(false);
    private volatile RateLimiter currentRateLimiter;
 
    @PostConstruct
    public void initRampUp() {
        // Start at 20% of normal rate, ramp up to 120% over 5 minutes
        currentRateLimiter = RateLimiter.create(NORMAL_RATE_PER_SECOND * 0.20);
 
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(
            this::increaseRate,
            30,   // Start increasing after 30 seconds
            30,   // Increase every 30 seconds
            TimeUnit.SECONDS
        );
    }
 
    private void increaseRate() {
        double currentRate = currentRateLimiter.getRate();
        double targetRate = Math.min(NORMAL_RATE_PER_SECOND * 1.2, MAX_SENDGRID_RATE);
 
        if (currentRate < targetRate) {
            double newRate = Math.min(currentRate * 1.5, targetRate);
            currentRateLimiter = RateLimiter.create(newRate);
            log.info("Consumer ramp-up: rate increased to {}/second", newRate);
 
            if (newRate >= targetRate) {
                rampUpComplete.set(true);
                log.info("Consumer ramp-up complete: steady state rate {}/second", newRate);
            }
        }
    }
 
    @KafkaListener(topics = "notification-commands", groupId = "notification-processor")
    public void handleNotification(NotificationCommand command, Acknowledgment ack) {
        // Acquire rate limit token before processing
        // This naturally throttles the consumer during ramp-up
        currentRateLimiter.acquire();
 
        try {
            notificationSender.send(command);
            ack.acknowledge();
        } catch (SendGridRateLimitException e) {
            // Even with our rate limiter, hit SendGrid limit
            // Back off and do not ack (redelivery after visibility timeout)
            log.warn("SendGrid rate limit hit, reducing consumer rate");
            currentRateLimiter = RateLimiter.create(currentRateLimiter.getRate() * 0.5);
        }
    }
}

# Step 1: List all consumer groups
kafka-consumer-groups.sh --bootstrap-server kafka:9092 --list
 
# Step 2: Identify zombie (inactive) groups
kafka-consumer-groups.sh --bootstrap-server kafka:9092 --all-groups --describe \
  | awk '$6 <mark class="obsidian-highlight"> "-" || $6 </mark> "0"' | grep -v "^GROUP"
# Groups with STATE=Empty or Dead and no active members are zombies
 
# Step 3: Archive the group's offsets before deletion
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --describe --group analytics-processor > /archive/analytics-processor-offsets.txt
 
# Step 4: Delete zombie consumer groups
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --delete --group analytics-processor
 
# Prevention: Consumer group naming convention
# Format: <service>.<environment>.<version>
# analytics-v1.prod.2025-03  (includes version and date)
# analytics-v2.prod.2025-11  (new service = new name, can never conflict)
 
// Prevention: Code-level safeguard
@Configuration
public class ConsumerGroupValidation {
 
    @Value("${spring.application.name}")
    private String serviceName;
 
    @Value("${spring.application.version}")
    private String version;
 
    @Value("${spring.profiles.active}")
    private String environment;
 
    @Bean
    public String consumerGroupId() {
        // Auto-generate consumer group from service identity
        // Cannot reuse old group by accident
        return String.format("%s.%s.%s", serviceName, environment, version);
    }
}

# Alert 30 days before expiry
# Using Prometheus blackbox exporter for certificate monitoring
 
- job_name: 'kafka-cert-check'
  metrics_path: /probe
  params:
    module: [tcp_connect]
  static_configs:
    - targets:
      - kafka-broker-1:9093
      - kafka-broker-2:9093
      - kafka-broker-3:9093
  relabel_configs:
    - source_labels: [__address__]
      target_label: __param_target
    - source_labels: [__param_target]
      target_label: instance
    - target_label: __address__
      replacement: blackbox-exporter:9115
 
# Alert rule
- alert: KafkaCertificateExpiringSoon
  expr: probe_ssl_earliest_cert_expiry - time() < 30 * 24 * 3600
  labels:
    severity: warning
  annotations:
    summary: "Kafka TLS certificate expires in {{ $value | humanizeDuration }}"
 
- alert: KafkaCertificateExpiryCritical
  expr: probe_ssl_earliest_cert_expiry - time() < 7 * 24 * 3600
  labels:
    severity: critical
  annotations:
    summary: "CRITICAL: Kafka TLS certificate expires in {{ $value | humanizeDuration }}"

// Producer-side graceful handling of certificate expiry
@Component
public class ResilientKafkaProducer {
 
    @Value("${kafka.bootstrap.servers}")
    private String bootstrapServers;
 
    private volatile KafkaProducer<String, byte[]> producer;
 
    public void publishWithFallback(String topic, String key, Object event) {
        try {
            producer.send(new ProducerRecord<>(topic, key, serialize(event)));
        } catch (AuthenticationException | SSLException e) {
            // Certificate-related failure - save to outbox for later
            log.error("Kafka auth failure (possible cert expiry): {}", e.getMessage());
            outboxRepository.save(OutboxEntry.builder()
                .topic(topic)
                .key(key)
                .payload(serialize(event))
                .failureReason("KAFKA_AUTH_FAILURE")
                .createdAt(Instant.now())
                .build());
            // Alert immediately - this needs human intervention
            alertingService.sendCritical("Kafka authentication failure - possible certificate expiry");
        }
    }
}

// Financial state machine with strict ordering enforcement
 
@Service
public class LoanStateMachineConsumer {
 
    @KafkaListener(
        topics = "loan-events",
        groupId = "loan-state-machine",
        containerFactory = "orderedConsumerFactory"  // BATCH mode, manual ack
    )
    public void processLoanEvents(
            List<ConsumerRecord<String, LoanEvent>> records,
            Acknowledgment ack) {
 
        // Process sequentially - NEVER async within a batch
        // Ordering within a partition is guaranteed by Kafka
        // Sequential processing ensures we respect that ordering
        for (ConsumerRecord<String, LoanEvent> record : records) {
            try {
                processInOrder(record.value(), record.offset());
            } catch (InvalidStateTransitionException e) {
                // State transition guard violated - this is a serious consistency error
                log.error("CRITICAL: Invalid state transition loan={}, event={}, currentState={}",
                    record.value().getLoanId(),
                    record.value().getType(),
                    e.getCurrentState());
 
                // Move to investigation DLQ with full context
                Map<String, String> headers = Map.of(
                    "violation-type", "INVALID_STATE_TRANSITION",
                    "current-state", e.getCurrentState(),
                    "attempted-transition", record.value().getType().name(),
                    "offset", String.valueOf(record.offset()),
                    "partition", String.valueOf(record.partition())
                );
                dlqPublisher.sendWithHeaders("loan-events.state-violations", record, headers);
                // Continue processing other loans - do not block on one invalid transition
            }
        }
 
        ack.acknowledge();
    }
 
    @Transactional
    private void processInOrder(LoanEvent event, long offset) {
        Loan loan = loanRepository.findById(event.getLoanId())
            .orElseThrow(() -> new LoanNotFoundException(event.getLoanId()));
 
        // Version-based idempotency: if we've already processed this or a later event, skip
        if (loan.getLastProcessedOffset() >= offset) {
            log.info("Already processed offset {} for loan {}, skipping", offset, event.getLoanId());
            return;
        }
 
        // Validate transition BEFORE applying
        stateMachine.validateTransition(loan.getStatus(), event.getType());
 
        // Apply transition and record offset
        loan.apply(event);
        loan.setLastProcessedOffset(offset);
        loanRepository.save(loan);
    }
}

// Fix 1: Producer-level circuit breaker + async fallback
@Service
public class BackpressureAwareOrderPublisher {
 
    private final CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("kafka-order-publisher");
    private final Bulkhead bulkhead = Bulkhead.of("kafka-publish",
        BulkheadConfig.custom().maxConcurrentCalls(50).build());
 
    public void publishOrderEvent(OrderEvent event) {
        try {
            // Bulkhead prevents thread pool exhaustion on the CALLER side
            // Max 50 concurrent publish attempts
            Bulkhead.decorateRunnable(bulkhead, () ->
                CircuitBreaker.decorateRunnable(circuitBreaker, () ->
                    kafkaTemplate.send("order-events", event.getOrderId(), event)
                        .get(2, TimeUnit.SECONDS)  // Never block longer than 2s
                ).run()
            ).run();
 
        } catch (BulkheadFullException e) {
            // 50 threads already waiting - Kafka is backed up
            // Save to outbox immediately, don't queue more
            log.warn("Kafka publish bulkhead full - routing to outbox");
            outboxRepository.save(OutboxEntry.from(event));
 
        } catch (CallNotPermittedException e) {
            // Circuit is open - Kafka unavailable
            // Save to outbox, circuit will close when Kafka recovers
            log.warn("Kafka circuit open - routing to outbox");
            outboxRepository.save(OutboxEntry.from(event));
 
        } catch (TimeoutException e) {
            // Individual send timeout - slow consumer (backpressure signal)
            // Route to outbox and open circuit if this is frequent
            circuitBreaker.onError(2, TimeUnit.SECONDS, e);
            outboxRepository.save(OutboxEntry.from(event));
        }
    }
}
 
// Fix 2: Consumer circuit breaker - slow down consumption when DB is slow
// This signals backpressure upstream rather than building up retry storms
 
@KafkaListener(topics = "order-events")
public void processOrder(OrderEvent event, Acknowledgment ack) {
    try {
        CircuitBreaker.decorateRunnable(dbCircuitBreaker, () -> {
            orderRepository.save(Order.from(event)); // The slow DB call
        }).run();
        ack.acknowledge();
 
    } catch (CallNotPermittedException e) {
        // DB circuit is open - don't ack, let message be redelivered
        // This naturally reduces consumer throughput when DB is unavailable
        // Consumer lag builds up in Kafka (which has infinite buffer) rather than
        // cascading into producer thread pool exhaustion
        log.warn("DB circuit open - not processing, will retry when DB recovers");
        Thread.sleep(5000); // Back off before next poll
    }
}

// Active-active is fundamentally hard. Choose your strategy based on data type:
 
// Strategy 1: Last Write Wins (LWW) with vector clocks (not wall clock)
// Use for: User preferences, profile data that users understand can conflict
@Service
public class ProfileConflictResolver {
 
    public UserProfile resolve(UserProfile local, UserProfile remote) {
        // Use vector clocks, not wall-clock time
        // Vector clock comparison is causally correct; wall clock is not
        VectorClock localClock = local.getVectorClock();
        VectorClock remoteClock = remote.getVectorClock();
 
        if (localClock.happensBefore(remoteClock)) {
            return remote; // Remote is causally after local - remote wins
        } else if (remoteClock.happensBefore(localClock)) {
            return local;  // Local is causally after remote - local wins
        } else {
            // Concurrent writes - this is the true conflict
            // Strategy: merge non-conflicting fields, flag conflicting ones for user resolution
            return mergeWithUserNotification(local, remote);
        }
    }
}
 
// Strategy 2: Region-Owned Data (recommended for most cases)
// Each user is "owned" by one region - only that region accepts writes for their data
// Other regions redirect writes to the owning region
// No conflicts possible - at the cost of latency for cross-region writes
 
@Service
public class RegionOwnershipRouter {
    public void updateProfile(String userId, ProfileUpdate update) {
        String owningRegion = userOwnershipRegistry.getOwningRegion(userId);
        if (owningRegion.equals(currentRegion)) {
            profileService.applyUpdate(userId, update);
        } else {
            // Proxy to owning region (accept slightly higher latency)
            regionRouter.forward(owningRegion, userId, update);
        }
    }
}
 
// Strategy 3: CRDT (Conflict-free Replicated Data Types)
// For data that naturally merges: shopping carts, feature flags, counters
// Mathematical properties guarantee convergence regardless of update order
 
@Service
public class CartCRDTConsumer {
    // G-Set (grow-only set) for cart items - no conflicts possible
    // Adding the same item twice = idempotent (set semantics)
    // Removing items requires 2P-Set (two-phase set) approach
}