java-topology/defects/pulsar/unit/PulsarPersistentTopicReplicationTest.java

package unit;

import java.util.*;

/**
 * pulsar-0003 + pulsar-0004: PersistentTopic replicationClusters / shadowTopics List → HashSet
 *
 * Demonstrates that List.contains() inside a replicator forEach loop is O(R×C),
 * while HashSet.contains() is O(R + C).
 *
 * Compile: javac -d . PulsarPersistentTopicReplicationTest.java
 * Run:     java unit.PulsarPersistentTopicReplicationTest
 */
public class PulsarPersistentTopicReplicationTest {

    // --- pulsar-0003: checkReplication() replicationClusters List.contains() ---

    // SLOW: mirrors PersistentTopic.checkReplication() with List<String>
    static int slowCheckReplication(List<String> replicators, List<String> configuredClusters) {
        int toRemove = 0;
        for (String cluster : replicators) {
            if (!configuredClusters.contains(cluster)) {  // O(C) per replicator
                toRemove++;
            }
        }
        return toRemove;
    }

    // FAST: fixed version — wrap in HashSet first
    static int fastCheckReplication(List<String> replicators, List<String> configuredClusters) {
        Set<String> clusterSet = new HashSet<>(configuredClusters);  // O(C) once
        int toRemove = 0;
        for (String cluster : replicators) {
            if (!clusterSet.contains(cluster)) {  // O(1) per replicator
                toRemove++;
            }
        }
        return toRemove;
    }

    // --- pulsar-0004: checkShadowReplication() shadowTopics List.contains() ---

    // SLOW: mirrors PersistentTopic.checkShadowReplication() with List<String>
    static int slowCheckShadow(List<String> shadowReplicators, List<String> configuredShadowTopics) {
        int toRemove = 0;
        for (String topic : shadowReplicators) {
            if (!configuredShadowTopics.contains(topic)) {  // O(S) per replicator
                toRemove++;
            }
        }
        return toRemove;
    }

    // FAST: fixed version
    static int fastCheckShadow(List<String> shadowReplicators, List<String> configuredShadowTopics) {
        Set<String> topicSet = new HashSet<>(configuredShadowTopics);  // O(S) once
        int toRemove = 0;
        for (String topic : shadowReplicators) {
            if (!topicSet.contains(topic)) {  // O(1)
                toRemove++;
            }
        }
        return toRemove;
    }

    static long bench(Runnable r, int iters) {
        for (int i = 0; i < 3; i++) r.run();
        long t0 = System.nanoTime();
        for (int i = 0; i < iters; i++) r.run();
        return System.nanoTime() - t0;
    }

    public static void main(String[] args) {
        System.out.println("pulsar-0003/0004: PersistentTopic replication List.contains → HashSet");
        System.out.println("=".repeat(70));

        int[] sizes = {10, 50, 100};
        int iters = 2000;
        boolean allPass = true;

        System.out.println("\n[pulsar-0003] checkReplication() — replicationClusters List.contains");
        for (int n : sizes) {
            List<String> configured = new ArrayList<>();
            for (int i = 0; i < n; i++) configured.add("cluster-" + i);

            // Active replicators: same as configured + some orphans not in configured
            List<String> replicators = new ArrayList<>(configured);
            for (int i = n; i < n + n / 2; i++) replicators.add("cluster-" + i);

            int slowRes = slowCheckReplication(replicators, configured);
            int fastRes = fastCheckReplication(replicators, configured);
            boolean correct = slowRes == fastRes && slowRes == n / 2;
            if (!correct) allPass = false;

            long slowNs = bench(() -> slowCheckReplication(replicators, configured), iters);
            long fastNs = bench(() -> fastCheckReplication(replicators, configured), iters);
            double ratio = (double) slowNs / fastNs;

            System.out.printf("  N=%-4d  slow=%7.3f ms  fast=%7.3f ms  ratio=%5.1fx  orphans=%d  %s%n",
                n,
                slowNs / 1_000_000.0 / iters,
                fastNs / 1_000_000.0 / iters,
                ratio, slowRes,
                correct ? "PASS" : "FAIL");
        }

        System.out.println("\n[pulsar-0004] checkShadowReplication() — shadowTopics List.contains");
        for (int n : sizes) {
            List<String> configured = new ArrayList<>();
            for (int i = 0; i < n; i++) configured.add("shadow-topic-" + i);

            List<String> replicators = new ArrayList<>(configured);
            for (int i = n; i < n + n / 2; i++) replicators.add("shadow-topic-" + i);

            int slowRes = slowCheckShadow(replicators, configured);
            int fastRes = fastCheckShadow(replicators, configured);
            boolean correct = slowRes == fastRes && slowRes == n / 2;
            if (!correct) allPass = false;

            long slowNs = bench(() -> slowCheckShadow(replicators, configured), iters);
            long fastNs = bench(() -> fastCheckShadow(replicators, configured), iters);
            double ratio = (double) slowNs / fastNs;

            System.out.printf("  N=%-4d  slow=%7.3f ms  fast=%7.3f ms  ratio=%5.1fx  orphans=%d  %s%n",
                n,
                slowNs / 1_000_000.0 / iters,
                fastNs / 1_000_000.0 / iters,
                ratio, slowRes,
                correct ? "PASS" : "FAIL");
        }

        System.out.println("\n" + "=".repeat(70));
        System.out.println(allPass ? "ALL PASS" : "SOME FAILED");
        if (!allPass) System.exit(1);
    }
}