package com.example.demo;

import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.common.policies.data.ConsumerStats;
import org.slf4j.LoggerFactory;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;

/**
 * PIP-470 Key_Shared Hot Key Overflow Scenario Verification Test
 *
 * <h2>Background</h2>
 * In Key_Shared subscriptions, a single Consumer failure (thread hang / downstream timeout) causes its
 * permits to drop to zero, and messages for its assigned Keys flood into the Replay queue. When the
 * Replay queue grows to the effectiveLookAheadLimit, Normal Read is completely blocked, causing all
 * other normal Keys to starve — going from full-speed consumption to "global starvation".
 *
 * <h2>Fault Simulation Method</h2>
 * <pre>
 *   After the faulty Consumer enters stuck mode:
 *   1. Call consumer.pause() — notify the client to stop requesting new messages from the Broker
 *   2. Stop calling consumer.receive() — thread hangs, simulating a real business hang
 *   → Client receiverQueue fills up, no more Flow requests are sent
 *   → Broker-side permits drop to zero
 *   → Messages for this consumer's keys cannot be dispatched → enter Replay queue
 *   → Replay queue grows to lookAheadLimit → Normal Read is blocked → global starvation
 * </pre>
 *
 * <h2>Configurable Parameters (via -D)</h2>
 * <pre>
 *   -Dpulsar.url              Pulsar cluster address, default: pulsar://xxxx:6650
 *   -Dpulsar.admin.url        Pulsar Admin HTTP address, default: http://xxxx:8080
 *   -Dpulsar.topic            Topic name (non-partitioned), default: persistent://sit/yuechu1/test-keyshared-hotkey
 *   -Dsub                     Subscription name, default: sub-hotkey-test
 *   -DconsumerCount           Number of consumers, default: 5
 *   -DkeyCount                Number of distinct keys, default: 50
 *   -DproduceTps              Producer total TPS, default: 2500
 *   -DmsgSize                 Message body size (bytes), default: 256
 *   -DwarmupSec               Warmup duration (seconds), default: 30
 *   -DstuckDurationSec        Fault duration (seconds), default: 180
 *   -DreceiverQueueSize       Consumer receiverQueueSize, default: 100 (small value accelerates permit drain)
 *   -DprocessDelayMs          Normal Consumer per-message processing delay (simulates business latency), default: 15
 *   -DlogLevel                Log level, default: WARN
 * </pre>
 *
 * <h2>Usage Examples</h2>
 * <pre>
 *   java -jar hotkey-test.jar
 *   java -DconsumerCount=5 -DkeyCount=50 -DproduceTps=2500 -DreceiverQueueSize=100 -jar hotkey-test.jar
 * </pre>
 */
public class KeySharedHotKeyBlockingTest {

    private static final org.slf4j.Logger log = LoggerFactory.getLogger(KeySharedHotKeyBlockingTest.class);
    private static final SimpleDateFormat SDF = new SimpleDateFormat("HH:mm:ss.SSS");

    // ==================== Configuration ====================
    static final String SERVICE_URL = prop("pulsar.url", "pulsar://xxxx:6650");
    static final String ADMIN_URL = prop("pulsar.admin.url", "http://xxxx:8080");
    static final String TOPIC = prop("pulsar.topic", "persistent://sit/yuechu1/test-keyshared-hotkey");
    static final String SUB = prop("sub", "sub-hotkey-test");

    static final int CONSUMER_COUNT = intProp("consumerCount", 5);
    static final int KEY_COUNT = intProp("keyCount", 50);
    static final int PRODUCE_TPS = intProp("produceTps", 2500);
    static final int MSG_SIZE = intProp("msgSize", 256);
    static final int WARMUP_SEC = intProp("warmupSec", 30);
    static final int STUCK_DURATION_SEC = intProp("stuckDurationSec", 1200);
    // Default 100: after fault injection, permits drain to zero in ~0.2s, accelerating Replay accumulation
    static final int RECEIVER_QUEUE_SIZE = intProp("receiverQueueSize", 100);
    static final int PROCESS_DELAY_MS = intProp("processDelayMs", 15);
    static final String LOG_LEVEL = prop("logLevel", "WARN");

    // effectiveLookAheadLimit estimate (for logging; actual value is determined by Broker config)
    static final int LOOK_AHEAD_PER_CONSUMER = intProp("lookAheadPerConsumer", 2000);
    static final int LOOK_AHEAD_PER_SUB = intProp("lookAheadPerSub", 20000);
    static final int EFFECTIVE_LOOK_AHEAD = Math.min(
            LOOK_AHEAD_PER_CONSUMER * CONSUMER_COUNT, LOOK_AHEAD_PER_SUB);

    // ==================== Statistics ====================
    static final AtomicLong sentOk = new AtomicLong();
    static final AtomicLong sentFail = new AtomicLong();
    static final AtomicLong totalAcked = new AtomicLong();
    static final AtomicLong healthyAcked = new AtomicLong();
    static final AtomicLong stuckWarmupAcked = new AtomicLong();

    // Per-consumer independent counters
    static final ConcurrentHashMap<String, AtomicLong> perConsumerAcked = new ConcurrentHashMap<>();

    // Time-window statistics (for detecting TPS drops)
    // Buffer size covers the entire test duration to avoid warmup data eviction
    static final int HISTORY_MAX_SIZE = WARMUP_SEC + STUCK_DURATION_SEC + 120;
    static final ConcurrentLinkedQueue<long[]> healthyTpsHistory = new ConcurrentLinkedQueue<>();

    // Baseline TPS captured in real-time during warmup (avoids retrospective lookup from bounded buffer)
    static volatile double capturedBaselineTps = -1;

    static volatile boolean running = true;
    static volatile boolean stuckMode = false;
    static final long START_TIME = System.currentTimeMillis();

    // Reference to stuck consumer, used for calling pause() from the main thread
    static volatile Consumer<byte[]> stuckConsumerRef = null;

    // ==================== Entry Point ====================

    public static void main(String[] args) throws Exception {
        ((Logger) LoggerFactory.getLogger("org.apache.pulsar")).setLevel(parseLogLevel(LOG_LEVEL));
        ((Logger) LoggerFactory.getLogger("io.netty")).setLevel(parseLogLevel(LOG_LEVEL));

        printConfig();

        // Recreate topic (non-partitioned)
        recreateTopic();

        PulsarClient client = buildClient();

        // Create multiple Consumers
        List<Consumer<byte[]>> consumers = new ArrayList<>();
        for (int i = 0; i < CONSUMER_COUNT; i++) {
            String name = "consumer-" + i;
            Consumer<byte[]> consumer = buildConsumer(client, name);
            consumers.add(consumer);
            perConsumerAcked.put(name, new AtomicLong());
            System.out.printf("[Main] Consumer '%s' created%n", name);
        }

        // Consumer-0 will become the stuck consumer after warmup
        String stuckConsumerName = "consumer-0";
        stuckConsumerRef = consumers.get(0);
        System.out.printf("[Main] After %ds warmup, '%s' will simulate failure (pause + stop receive)%n",
                WARMUP_SEC, stuckConsumerName);

        // Create Producer
        Producer<byte[]> producer = buildProducer(client);

        // Start consume threads
        ExecutorService consumePool = Executors.newFixedThreadPool(CONSUMER_COUNT, r -> {
            Thread t = new Thread(r);
            t.setDaemon(true);
            return t;
        });
        for (int i = 0; i < CONSUMER_COUNT; i++) {
            final Consumer<byte[]> c = consumers.get(i);
            final String name = "consumer-" + i;
            consumePool.submit(() -> consumeLoop(c, name, name.equals(stuckConsumerName)));
        }

        // Start Producer thread
        Thread produceThread = new Thread(() -> produceLoop(producer), "producer-thread");
        produceThread.setDaemon(true);
        produceThread.start();

        // Metrics printing (every second)
        ScheduledExecutorService metrics = Executors.newSingleThreadScheduledExecutor();
        metrics.scheduleAtFixedRate(new MetricsPrinter(), 1, 1, TimeUnit.SECONDS);

        System.out.printf("%n[Main] All components started. Test flow: warmup %ds -> fault injection %ds -> auto stop%n%n",
                WARMUP_SEC, STUCK_DURATION_SEC);

        // ==================== Phase 1: Warmup ====================
        System.out.printf("[Main] === Phase 1/2: Warmup (%ds) — all Consumers ack normally ===%n", WARMUP_SEC);
        Thread.sleep(WARMUP_SEC * 1000L);

        // ==================== Phase 2: Fault Injection ====================
        long stuckStartTime = System.currentTimeMillis();

        // Capture baseline TPS from warmup data before it could possibly be evicted
        capturedBaselineTps = computeBaselineTps(stuckStartTime);
        System.out.printf("[Main] Captured baseline TPS at warmup end: %.1f msg/s%n", capturedBaselineTps);

        // First call pause(): notify the client to stop sending Flow requests to the Broker
        System.out.printf("%n[Main] === Phase 2/2: Fault Injection ===%n");
        System.out.printf("[Main] Calling consumer-0.pause() — stop Flow requests%n");
        stuckConsumerRef.pause();

        // Then set stuckMode to make the consume thread stop calling receive()
        stuckMode = true;
        System.out.printf("[Main] Set stuckMode=true — consumer-0 stops receive(), simulating thread hang%n");
        System.out.printf("[Main] Fault duration %ds, estimated Replay fill time ~%.1f seconds%n",
                STUCK_DURATION_SEC,
                (double) EFFECTIVE_LOOK_AHEAD / (KEY_COUNT / CONSUMER_COUNT * PRODUCE_TPS / KEY_COUNT));

        // Start Broker-side metrics collection thread (every 10 seconds, collecting consumer permits)
        ScheduledExecutorService brokerMetrics = Executors.newSingleThreadScheduledExecutor();
        brokerMetrics.scheduleAtFixedRate(() -> collectBrokerMetrics(stuckStartTime), 5, 10, TimeUnit.SECONDS);

        // Fault duration phase
        Thread.sleep(STUCK_DURATION_SEC * 1000L);

        // ==================== Test Complete ====================
        running = false;
        System.out.println("\n[Main] Test time reached, stopping...");

        Thread.sleep(3000);
        metrics.shutdownNow();
        brokerMetrics.shutdownNow();
        consumePool.shutdownNow();
        producer.close();
        for (Consumer<byte[]> c : consumers) {
            try { c.close(); } catch (Exception ignore) {}
        }
        client.close();

        printSummary(stuckConsumerName, stuckStartTime);
        destroyTopic();
    }

    // ==================== Broker-Side Metrics Collection ====================

    /**
     * Collect each consumer's availablePermits and the subscription's msgBacklog via Admin API,
     * to confirm whether the stuck consumer's permits have drained to zero and whether the Replay queue is growing.
     */
    static void collectBrokerMetrics(long stuckStartTime) {
        try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(ADMIN_URL).build()) {
            TopicStats stats = admin.topics().getStats(TOPIC);
            long elapsed = (System.currentTimeMillis() - stuckStartTime) / 1000;

            SubscriptionStats subStats = stats.getSubscriptions().get(SUB);
            if (subStats == null) {
                System.out.printf("[BrokerMetrics][T+%ds] Subscription '%s' not found%n", elapsed, SUB);
                return;
            }

            long backlog = subStats.getMsgBacklog();
            long unackedMessages = subStats.getUnackedMessages();
            int replayMsgs = (int) subStats.getMsgRateRedeliver(); // Redelivery rate (approximates replay activity)

            StringBuilder sb = new StringBuilder();
            sb.append(String.format("[BrokerMetrics][T+%ds] backlog=%d unacked=%d redeliveryRate=%.1f | permits: ",
                    elapsed, backlog, unackedMessages, subStats.getMsgRateRedeliver()));

            List<? extends ConsumerStats> consumerStatsList = subStats.getConsumers();
            if (consumerStatsList != null) {
                for (ConsumerStats cs : consumerStatsList) {
                    sb.append(String.format("%s=%d ", cs.getConsumerName(), cs.getAvailablePermits()));
                }
            }

            System.out.println(sb.toString().trim());
        } catch (Exception e) {
            // Collection failure does not affect the test
            if (running) {
                System.err.printf("[BrokerMetrics] Collection failed: %s%n", e.getMessage());
            }
        }
    }

    // ==================== Topic Management ====================

    static void recreateTopic() {
        System.out.printf("[Admin] Preparing to recreate non-partitioned topic: %s%n", TOPIC);
        try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(ADMIN_URL).build()) {
            try {
                admin.topics().delete(TOPIC, true);
                System.out.printf("[Admin] Deleted old topic: %s%n", TOPIC);
            } catch (PulsarAdminException.NotFoundException e) {
                System.out.printf("[Admin] Topic does not exist, no need to delete: %s%n", TOPIC);
            }
            try {
                admin.topics().deletePartitionedTopic(TOPIC, true);
            } catch (PulsarAdminException.NotFoundException ignore) {}

            Thread.sleep(2000);

            try {
                admin.topics().createNonPartitionedTopic(TOPIC);
                System.out.printf("[Admin] Created non-partitioned topic: %s%n", TOPIC);
            } catch (Exception e) {
                System.out.printf("[Admin] Exception creating topic (may have been auto-created): %s%n", e.getMessage());
            }

            Thread.sleep(1000);
        } catch (Exception e) {
            System.err.printf("[Admin] Failed to recreate topic: %s%n", e.getMessage());
            e.printStackTrace();
            System.exit(1);
        }
    }

    static void destroyTopic() {
        System.out.printf("[Admin] Destroying topic: %s%n", TOPIC);
        try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(ADMIN_URL).build()) {
            admin.topics().delete(TOPIC, true);
            System.out.printf("[Admin] Topic destroyed: %s%n", TOPIC);
        } catch (PulsarAdminException.NotFoundException e) {
            System.out.printf("[Admin] Topic already does not exist: %s%n", TOPIC);
        } catch (Exception e) {
            System.err.printf("[Admin] Failed to destroy topic (can be cleaned up manually): %s%n", e.getMessage());
        }
    }

    // ==================== Producer Loop ====================

    static void produceLoop(Producer<byte[]> producer) {
        String[] keys = new String[KEY_COUNT];
        for (int i = 0; i < KEY_COUNT; i++) {
            keys[i] = "key-" + i;
        }
        ThreadLocalRandom rng = ThreadLocalRandom.current();
        RateLimiter limiter = new RateLimiter(PRODUCE_TPS);
        Semaphore semaphore = new Semaphore(512);

        System.out.printf("[Producer] Starting to send: %d keys, %d TPS, %d bytes/msg%n",
                KEY_COUNT, PRODUCE_TPS, MSG_SIZE);

        long seq = 0;
        while (running) {
            limiter.acquire(1);

            String key = keys[(int) (seq % KEY_COUNT)];
            byte[] payload = new byte[MSG_SIZE];
            rng.nextBytes(payload);
            seq++;

            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }

            producer.newMessage().key(key).value(payload).sendAsync()
                    .thenRun(() -> {
                        sentOk.incrementAndGet();
                        semaphore.release();
                    })
                    .exceptionally(ex -> {
                        sentFail.incrementAndGet();
                        semaphore.release();
                        if (sentFail.get() % 1000 == 1) {
                            System.err.printf("[Producer] Send failed: %s%n", ex.getMessage());
                        }
                        return null;
                    });
        }

        try {
            semaphore.acquire(512);
            semaphore.release(512);
        } catch (InterruptedException ignore) {}

        System.out.printf("[Producer] Stopped: sentOk=%d, sentFail=%d%n", sentOk.get(), sentFail.get());
    }

    // ==================== Consumer Loop ====================

    static void consumeLoop(Consumer<byte[]> consumer, String name, boolean isStuckCandidate) {
        System.out.printf("[%s] Starting to consume (isStuckCandidate=%s)%n", name, isStuckCandidate);

        while (running) {
            try {
                // ===== Stuck mode: completely stop receive(), simulating a real business hang =====
                // Key point: must NOT continue calling receive()! Otherwise the client will keep
                //            consuming from the internal queue, send Flow requests to replenish
                //            permits, and the Broker will think the consumer is healthy — messages
                //            won't enter the Replay queue.
                // Correct approach: hang the thread + consumer.pause(), let receiverQueue fill up,
                //                   client stops Flow -> Broker permits drain to zero -> messages
                //                   enter Replay queue.
                if (isStuckCandidate && stuckMode) {
                    System.out.printf("[%s] *** HANG: thread suspended, no more receive(), waiting for Broker permits to drain ***%n", name);
                    // Hang until the test ends
                    while (running) {
                        Thread.sleep(5000);
                    }
                    return;
                }

                Message<byte[]> msg = consumer.receive(2, TimeUnit.SECONDS);
                if (msg == null) continue;

                // Normal mode: process and ack
                if (PROCESS_DELAY_MS > 0) {
                    Thread.sleep(PROCESS_DELAY_MS);
                }
                consumer.acknowledge(msg);

                if (isStuckCandidate) {
                    stuckWarmupAcked.incrementAndGet();
                } else {
                    healthyAcked.incrementAndGet();
                }
                totalAcked.incrementAndGet();
                perConsumerAcked.get(name).incrementAndGet();
            } catch (Exception e) {
                if (running) {
                    System.err.printf("[%s] Consume exception: %s%n", name, e.getMessage());
                }
            }
        }
    }

    // ==================== Builder Methods ====================

    static PulsarClient buildClient() throws Exception {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(SERVICE_URL)
                .operationTimeout(10, TimeUnit.SECONDS)
                .connectionsPerBroker(4)
                .build();
        log.warn("PulsarClient created: {}", SERVICE_URL);
        return client;
    }

    static Producer<byte[]> buildProducer(PulsarClient client) throws Exception {
        Producer<byte[]> producer = client.newProducer()
                .topic(TOPIC)
                .producerName("hotkey-producer-" + System.currentTimeMillis())
                .enableBatching(true)
                .batchingMaxPublishDelay(5, TimeUnit.MILLISECONDS)
                .batchingMaxMessages(500)
                .sendTimeout(10, TimeUnit.SECONDS)
                .maxPendingMessages(50000)
                .create();
        log.warn("Producer created on topic: {}", TOPIC);
        return producer;
    }

    static Consumer<byte[]> buildConsumer(PulsarClient client, String name) throws Exception {
        Consumer<byte[]> consumer = client.newConsumer()
                .topic(TOPIC)
                .subscriptionName(SUB)
                .subscriptionType(SubscriptionType.Key_Shared)
                .consumerName(name)
                .receiverQueueSize(RECEIVER_QUEUE_SIZE)
                .ackTimeout(0, TimeUnit.SECONDS)
                .subscribe();
        log.warn("Consumer '{}' created: type=Key_Shared, receiverQueueSize={}", name, RECEIVER_QUEUE_SIZE);
        return consumer;
    }

    // ==================== Per-Second Metrics ====================

    static class MetricsPrinter implements Runnable {
        long lastSent = 0, lastTotalAcked = 0, lastHealthyAcked = 0;

        @Override
        public void run() {
            long sent = sentOk.get();
            long tAcked = totalAcked.get();
            long hAcked = healthyAcked.get();

            long dSent = sent - lastSent;
            long dTotal = tAcked - lastTotalAcked;
            long dHealthy = hAcked - lastHealthyAcked;

            lastSent = sent;
            lastTotalAcked = tAcked;
            lastHealthyAcked = hAcked;

            long elapsed = (System.currentTimeMillis() - START_TIME) / 1000;
            String phase = stuckMode ? "STUCK" : "WARMUP";

            // Record healthy Consumer TPS history
            healthyTpsHistory.add(new long[]{elapsed, dHealthy});
            while (healthyTpsHistory.size() > HISTORY_MAX_SIZE) healthyTpsHistory.poll();

            // Per-consumer cumulative ack counts
            StringBuilder perConsumer = new StringBuilder();
            for (int i = 0; i < CONSUMER_COUNT; i++) {
                String cName = "consumer-" + i;
                AtomicLong counter = perConsumerAcked.get(cName);
                if (counter != null) {
                    if (perConsumer.length() > 0) perConsumer.append(" ");
                    perConsumer.append(String.format("c%d=%d", i, counter.get()));
                }
            }

            System.out.printf("[%s][%ds][%s] sendTps=%d ackTps=%d healthyTps=%d | " +
                            "cumul: sent=%d acked=%d healthy=%d warmup_c0=%d | [%s]%n",
                    SDF.format(new Date()), elapsed, phase,
                    dSent, dTotal, dHealthy,
                    sent, tAcked, hAcked, stuckWarmupAcked.get(),
                    perConsumer);
        }
    }

    // ==================== Result Summary ====================

    static void printSummary(String stuckConsumerName, long stuckStartTime) {
        long totalElapsed = (System.currentTimeMillis() - START_TIME) / 1000;
        long stuckElapsed = (System.currentTimeMillis() - stuckStartTime) / 1000;

        System.out.println("\n==================== PIP-470 Test Results ====================");
        System.out.println("  Topic:               " + TOPIC + "  (non-partitioned)");
        System.out.println("  Subscription:        " + SUB + "  (Key_Shared)");
        System.out.printf("  Consumer count:      %d (1 faulty)%n", CONSUMER_COUNT);
        System.out.printf("  Key count:           %d%n", KEY_COUNT);
        System.out.printf("  Producer TPS:        %d%n", PRODUCE_TPS);
        System.out.printf("  Message size:        %d bytes%n", MSG_SIZE);
        System.out.printf("  Process delay:       %d ms%n", PROCESS_DELAY_MS);
        System.out.printf("  receiverQueueSize:   %d%n", RECEIVER_QUEUE_SIZE);
        System.out.printf("  Est. lookAheadLimit: %d (per_consumer=%d x %d, per_sub=%d)%n",
                EFFECTIVE_LOOK_AHEAD, LOOK_AHEAD_PER_CONSUMER, CONSUMER_COUNT, LOOK_AHEAD_PER_SUB);
        System.out.printf("  Warmup duration:     %d seconds%n", WARMUP_SEC);
        System.out.printf("  Fault duration:      %d seconds (actual %d seconds)%n", STUCK_DURATION_SEC, stuckElapsed);
        System.out.printf("  Total run time:      %d seconds%n", totalElapsed);
        System.out.println();

        System.out.println("  [Producer]");
        System.out.printf("    Send success:      %d%n", sentOk.get());
        System.out.printf("    Send failure:      %d%n", sentFail.get());
        System.out.println();

        System.out.printf("  [Faulty Consumer: %s]%n", stuckConsumerName);
        System.out.printf("    Warmup phase ack:  %d%n", stuckWarmupAcked.get());
        System.out.printf("    Fault mode:        pause() + stop receive() (thread hang)%n");
        System.out.println();

        // Consumption rate statistics
        long totalSent = sentOk.get();
        long totalHealthy = healthyAcked.get();
        double consumeRate = totalSent > 0 ? 100.0 * (totalHealthy + stuckWarmupAcked.get()) / totalSent : 0;
        System.out.printf("  [Healthy Consumers x %d]%n", CONSUMER_COUNT - 1);
        System.out.printf("    Total ack count:   %d%n", totalHealthy);
        System.out.printf("    Consume rate:      %.2f%% (acked / sent = %d / %d)%n",
                consumeRate, totalHealthy + stuckWarmupAcked.get(), totalSent);
        System.out.println();

        System.out.println("  [Per-Consumer Ack Details]");
        for (int i = 0; i < CONSUMER_COUNT; i++) {
            String cName = "consumer-" + i;
            AtomicLong counter = perConsumerAcked.get(cName);
            long count = counter != null ? counter.get() : 0;
            String role = cName.equals(stuckConsumerName) ? " <- STUCK (pause + hang)" : "";
            System.out.printf("    %-15s : %10d%s%n", cName, count, role);
        }
        System.out.println();

        analyzeTpsTrend(stuckStartTime);

        System.out.println("========================================================");
    }

    /**
     * Analyze the TPS trend of healthy Consumers after fault injection.
     */
    static void analyzeTpsTrend(long stuckStartTime) {
        System.out.println("  [Healthy Consumer TPS Trend Analysis]");

        long warmupEnd = (stuckStartTime - START_TIME) / 1000;
        long baselineStart = Math.max(10, warmupEnd - 10);

        List<long[]> history = new ArrayList<>(healthyTpsHistory);
        if (history.isEmpty()) {
            System.out.println("    (Insufficient data for analysis)");
            return;
        }

        // Calculate baseline TPS
        long baselineSum = 0;
        int baselineCount = 0;
        for (long[] entry : history) {
            long t = entry[0];
            if (t >= baselineStart && t < warmupEnd) {
                baselineSum += entry[1];
                baselineCount++;
            }
        }
        double baselineTps = baselineCount > 0 ? (double) baselineSum / baselineCount : 0;
        System.out.printf("    Baseline TPS (end of warmup): %.1f msg/s%n", baselineTps);

        if (baselineTps <= 0) {
            System.out.println("    (Baseline TPS is 0, cannot analyze)");
            return;
        }

        // TPS changes after fault injection (one window every 30 seconds)
        int windowSec = 30;
        System.out.printf("    TPS changes after fault injection (every %ds window):%n", windowSec);

        int totalWindows = 0;
        int nearZeroWindows = 0;   // TPS < 20% of baseline (near starvation, residual from skipNextReplayToTriggerLookAhead)
        int degradedWindows = 0;   // TPS < 50% of baseline

        for (long windowStart = warmupEnd; windowStart < warmupEnd + STUCK_DURATION_SEC; windowStart += windowSec) {
            long windowEnd = windowStart + windowSec;
            long windowSum = 0;
            int windowCount = 0;
            for (long[] entry : history) {
                long t = entry[0];
                if (t >= windowStart && t < windowEnd) {
                    windowSum += entry[1];
                    windowCount++;
                }
            }
            if (windowCount > 0) {
                double windowTps = (double) windowSum / windowCount;
                double ratio = windowTps / baselineTps * 100;
                String bar = buildBar(ratio);
                System.out.printf("      T+%3d~%3ds: %6.1f msg/s (%5.1f%% of baseline) %s%n",
                        (int) (windowStart - warmupEnd),
                        (int) (windowEnd - warmupEnd),
                        windowTps, ratio, bar);
                totalWindows++;
                if (ratio < 20) nearZeroWindows++;
                if (ratio < 50) degradedWindows++;
            }
        }

        // Overall consumption rate
        long totalSent = sentOk.get();
        long totalConsumed = healthyAcked.get() + stuckWarmupAcked.get();
        double overallConsumeRate = totalSent > 0 ? 100.0 * totalConsumed / totalSent : 0;

        System.out.println();
        System.out.printf("    Overall consume rate: %.2f%% (consumed %d / sent %d)%n",
                overallConsumeRate, totalConsumed, totalSent);
        System.out.println();

        // Verdict logic:
        // Threshold note: TPS will not drop to absolute zero because Pulsar's
        // skipNextReplayToTriggerLookAhead mechanism allows 1 Normal Read per Replay round.
        // Therefore we use < 20% as the "near starvation" threshold.
        if (totalWindows == 0) {
            System.out.println("    Verdict: Insufficient data, cannot determine");
        } else if (nearZeroWindows > totalWindows / 2) {
            System.out.println("    Verdict: FAIL - Verified PIP-470 issue: single Consumer failure causes global starvation!");
            System.out.printf("           After fault injection, %d/%d windows had TPS < 20%% of baseline, normal Keys starved.%n",
                    nearZeroWindows, totalWindows);
            System.out.printf("           Overall consume rate only %.2f%%, out of %d messages sent, only %d consumed.%n",
                    overallConsumeRate, totalSent, totalConsumed);
            System.out.println("           Root cause: permits drained -> Replay queue full -> Normal Read blocked -> all Keys starved");
            System.out.println("           (Residual TPS from skipNextReplayToTriggerLookAhead allowing 1 Normal Read per round)");
        } else if (nearZeroWindows > 0) {
            System.out.println("    Verdict: WARNING - Partial blocking: faulty Consumer significantly affected other Keys' consumption");
            System.out.printf("           %d/%d windows had TPS < 20%% of baseline, intermittent starvation detected.%n",
                    nearZeroWindows, totalWindows);
        } else if (degradedWindows > totalWindows / 3) {
            System.out.println("    Verdict: WARNING - Significant consumption degradation: faulty Consumer caused backpressure effect");
            System.out.printf("           %d/%d windows had TPS < 50%% of baseline.%n",
                    degradedWindows, totalWindows);
        } else {
            System.out.println("    Verdict: PASS - Normal Key consumption largely unaffected: current Broker version may have mitigation");
            System.out.println("           or test parameters were insufficient to trigger Replay queue overflow.");
        }
    }

    static String buildBar(double pct) {
        int len = (int) Math.round(pct / 5);
        len = Math.max(0, Math.min(len, 20));
        StringBuilder sb = new StringBuilder("[");
        for (int i = 0; i < 20; i++) {
            sb.append(i < len ? '█' : '░');
        }
        sb.append(']');
        return sb.toString();
    }

    // ==================== Utilities ====================

    static class RateLimiter {
        final long ratePerSec;
        final int chunksPerSec;
        final long chunkIntervalNanos;
        final long permitsPerChunk;
        long chunkStart;
        long chunkPermitsLeft;

        RateLimiter(long ratePerSec) {
            this.ratePerSec = Math.max(1, ratePerSec);
            this.chunksPerSec = 100;
            this.chunkIntervalNanos = 1_000_000_000L / chunksPerSec;
            this.permitsPerChunk = Math.max(1, this.ratePerSec / chunksPerSec);
            this.chunkStart = System.nanoTime();
            this.chunkPermitsLeft = permitsPerChunk;
        }

        void acquire(int permits) {
            while (true) {
                if (chunkPermitsLeft >= permits) {
                    chunkPermitsLeft -= permits;
                    return;
                }
                long now = System.nanoTime();
                long elapsed = now - chunkStart;
                if (elapsed < chunkIntervalNanos) {
                    long sleepNanos = chunkIntervalNanos - elapsed;
                    if (sleepNanos > 1_000_000L) {
                        LockSupport.parkNanos(sleepNanos - 500_000L);
                    } else {
                        Thread.yield();
                    }
                } else {
                    chunkStart = now;
                    chunkPermitsLeft = permitsPerChunk;
                }
            }
        }
    }

    static String prop(String k, String def) { return System.getProperty(k, def); }
    static int intProp(String k, int def) { return Integer.parseInt(prop(k, String.valueOf(def))); }

    static Level parseLogLevel(String level) {
        switch (level.toUpperCase()) {
            case "DEBUG": return Level.DEBUG;
            case "INFO":  return Level.INFO;
            case "ERROR": return Level.ERROR;
            case "OFF":   return Level.OFF;
            default:      return Level.WARN;
        }
    }

    static String formatDuration(long totalSeconds) {
        long h = totalSeconds / 3600;
        long m = (totalSeconds % 3600) / 60;
        long s = totalSeconds % 60;
        if (h > 0) return String.format("%dh%02dm%02ds", h, m, s);
        if (m > 0) return String.format("%dm%02ds", m, s);
        return String.format("%ds", s);
    }

    static void printConfig() {
        int keysPerConsumer = KEY_COUNT / CONSUMER_COUNT;
        int tpsPerKey = PRODUCE_TPS / KEY_COUNT;
        int stuckKeyTps = keysPerConsumer * tpsPerKey;
        double permitDrainSec = (double) RECEIVER_QUEUE_SIZE / stuckKeyTps;
        double replayFillTimeSec = EFFECTIVE_LOOK_AHEAD / (double) stuckKeyTps;

        System.out.println("======== PIP-470 Key_Shared Hot Key Blocking Test ========");
        System.out.println("  pulsar.url            = " + SERVICE_URL);
        System.out.println("  pulsar.admin.url      = " + ADMIN_URL);
        System.out.println("  pulsar.topic          = " + TOPIC + "  (non-partitioned)");
        System.out.println("  sub                   = " + SUB);
        System.out.println("  subType               = Key_Shared");
        System.out.println();
        System.out.printf("  consumerCount         = %d%n", CONSUMER_COUNT);
        System.out.printf("  keyCount              = %d%n", KEY_COUNT);
        System.out.printf("  produceTps            = %d%n", PRODUCE_TPS);
        System.out.printf("  msgSize               = %d bytes%n", MSG_SIZE);
        System.out.printf("  processDelayMs        = %d ms%n", PROCESS_DELAY_MS);
        System.out.printf("  receiverQueueSize     = %d  <- small value accelerates permit drain%n", RECEIVER_QUEUE_SIZE);
        System.out.printf("  warmupSec             = %d%n", WARMUP_SEC);
        System.out.printf("  stuckDurationSec      = %d%n", STUCK_DURATION_SEC);
        System.out.println();
        System.out.println("  [Fault Simulation Method]");
        System.out.println("    consumer.pause() + stop receive() (thread hang)");
        System.out.println("    -> receiverQueue fills up -> stop Flow -> Broker permits drain to zero");
        System.out.println("    -> messages enter Replay queue -> once full, Normal Read is blocked");
        System.out.println();
        System.out.println("  [Theoretical Analysis]");
        System.out.printf("    Each Consumer handles ~%d Keys%n", keysPerConsumer);
        System.out.printf("    TPS per Key:               %d msg/s%n", tpsPerKey);
        System.out.printf("    Faulty Consumer key TPS:   ~%d msg/s%n", stuckKeyTps);
        System.out.printf("    Est. permit drain time:    ~%.1f seconds (receiverQueue=%d / %d msg/s)%n",
                permitDrainSec, RECEIVER_QUEUE_SIZE, stuckKeyTps);
        System.out.printf("    effectiveLookAheadLimit:   %d%n", EFFECTIVE_LOOK_AHEAD);
        System.out.printf("    Est. Replay queue fill:    ~%.1f seconds%n", replayFillTimeSec);
        System.out.printf("    *** Est. global starvation: ~%.0f seconds after fault injection ***%n",
                permitDrainSec + replayFillTimeSec);
        System.out.println();
        System.out.printf("  logLevel              = %s%n", LOG_LEVEL);
        System.out.println("=========================================================");
    }
}