Verify: migrate from Hazelcast to KiviDB

Count what is in each structure on both sides. On KiviDB, DBSIZE and INFO keyspace give the total and how many keys expire; count one map's keys by prefix with SCAN:

redis-cli -h kividb.internal -p 6380 -a "$KIVIDB_PASSWORD" INFO keyspace
redis-cli -h kividb.internal -p 6380 -a "$KIVIDB_PASSWORD" --scan --pattern 'customer:*' | wc -l

This program compares each map's size, then N random entries with their values and expiry, and the sizes of the other structures. It reuses Migrate.toHash, so build it with the copy program:

import com.hazelcast.client.HazelcastClient;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.core.EntryView;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.IMap;
import com.hazelcast.nio.serialization.genericrecord.GenericRecord;
import model.LegacyOrder;
import redis.clients.jedis.JedisPooled;
import redis.clients.jedis.params.ScanParams;
import redis.clients.jedis.resps.ScanResult;

import java.util.*;

// Compares counts and N random entries per IMap between Hazelcast and KiviDB, values and expiry.
public class Verify {
    static int bad = 0;

    public static void main(String[] args) {
        String hzAddress = System.getenv().getOrDefault("HZ_ADDRESS", "127.0.0.1:5701");
        String hzCluster = System.getenv().getOrDefault("HZ_CLUSTER", "dev");
        String kividbUrl = System.getenv().getOrDefault("KIVIDB_URL", "redis://127.0.0.1:6380");
        int n = Integer.parseInt(System.getenv().getOrDefault("SAMPLE", "500"));

        ClientConfig cc = new ClientConfig();
        cc.setClusterName(hzCluster);
        cc.getNetworkConfig().addAddress(hzAddress);
        HazelcastInstance hz = HazelcastClient.newHazelcastClient(cc);
        try (JedisPooled kv = new JedisPooled(java.net.URI.create(kividbUrl))) {
            checkMap(hz.getMap("customers"), kv, "customer:", n);
            checkMap(hz.getMap("sessions"), kv, "session:", n);
            checkMap(hz.getMap("stock"), kv, "stock:", n);
            checkMap(hz.getMap("legacy"), kv, "order:", n);
            same("multimap tags keys", hz.getMultiMap("tags").keySet().size(), countKeys(kv, "tags:*"));
            same("queue jobs", hz.getQueue("jobs").size(), kv.llen("jobs"));
            same("list events", hz.getList("events").size(), kv.llen("events"));
            same("set admins", hz.getSet("admins").size(), kv.scard("admins"));
            same("ringbuffer audit", hz.getRingbuffer("audit").size(), kv.xlen("audit"));
            same("replicated map config", hz.getReplicatedMap("config").size(), kv.hlen("config"));
            System.out.println("KiviDB DBSIZE " + kv.dbSize());
        }
        hz.shutdown();
        System.out.println(bad == 0 ? "OK: no mismatches" : "FAILED: " + bad + " mismatches");
        System.exit(bad == 0 ? 0 : 1);
    }

    static void checkMap(IMap<Object, Object> map, JedisPooled kv, String prefix, int n) {
        same("map " + map.getName() + " size", map.size(), countKeys(kv, prefix + "*"));
        List<Object> keys = new ArrayList<>(map.keySet());
        Collections.shuffle(keys);
        for (Object k : keys.subList(0, Math.min(n, keys.size()))) {
            String key = prefix + k;
            EntryView<Object, Object> ev = map.getEntryView(k);
            if (ev == null) continue;                              // expired since keySet()
            Object v = ev.getValue();
            boolean ok;
            if (v instanceof GenericRecord r) ok = Migrate.toHash(r).equals(kv.hgetAll(key));
            else if (v instanceof LegacyOrder o) ok = Map.of("id", Long.toString(o.id), "sku", o.sku, "qty", Integer.toString(o.qty)).equals(kv.hgetAll(key));
            else ok = v.toString().equals(kv.get(key));
            if (!ok) { bad++; System.out.println("value differs: " + key); continue; }
            long now = System.currentTimeMillis();
            long pttl = kv.pttl(key);
            boolean hzExpires = ev.getExpirationTime() != Long.MAX_VALUE;
            if (hzExpires != (pttl >= 0) || (hzExpires && Math.abs((ev.getExpirationTime() - now) - pttl) > 5000)) {
                bad++;
                System.out.printf("expiry differs: %s hazelcast=%d kividb_pttl=%d%n", key, ev.getExpirationTime() - now, pttl);
            }
        }
        System.out.printf("map %s: sampled %d%n", map.getName(), Math.min(n, keys.size()));
    }

    static long countKeys(JedisPooled kv, String pattern) {
        long c = 0;
        String cursor = ScanParams.SCAN_POINTER_START;
        ScanParams p = new ScanParams().match(pattern).count(1000);
        do {
            ScanResult<String> r = kv.scan(cursor, p);
            c += r.getResult().size();
            cursor = r.getCursor();
        } while (!cursor.equals(ScanParams.SCAN_POINTER_START));
        return c;
    }

    static void same(String what, long hzCount, long kvCount) {
        System.out.printf("%-28s hazelcast=%d kividb=%d%s%n", what, hzCount, kvCount, hzCount == kvCount ? "" : "  MISMATCH");
        if (hzCount != kvCount) bad++;
    }
}

map.keySet() brings every key to the client; on very large maps, sample from a few partitions' worth instead. Run it while writes are frozen, or expect small differences in maps that are changing.