Get the data out: migrate from Hazelcast to KiviDB

There is no file to export: Hazelcast's data lives in the members' memory (and in Persistence files on Enterprise, in a format only Hazelcast reads). The ways out all read through the Hazelcast API.

1. A client program that copies everything (the usual way)

The Java program below connects to the cluster as a client, iterates each structure, and writes it to KiviDB with pipelined commands. It uses the Hazelcast Java client (com.hazelcast:hazelcast, the same jar as the member) and Jedis 5 (redis.clients:jedis).

How it handles values:

• Compact (and Portable): the program does not need your classes. A value whose class isn't on the classpath arrives as a GenericRecord, and every field becomes a hash field.
• Java Serializable, IdentifiedDataSerializable, DataSerializable: the program needs your classes (and your DataSerializableFactory) on its classpath, and you write the mapping to fields, as for LegacyOrder below.
• Strings and numbers become strings.

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.multimap.MultiMap;
import com.hazelcast.nio.serialization.genericrecord.GenericRecord;
import com.hazelcast.ringbuffer.ReadResultSet;
import com.hazelcast.ringbuffer.Ringbuffer;
import model.LegacyOrder;
import redis.clients.jedis.JedisPooled;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.params.XAddParams;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

// Copies Hazelcast structures into KiviDB.
// Customer (Compact) is NOT on this program's classpath: its values arrive as GenericRecord.
// LegacyOrder (Java serialization) must be on the classpath.
public class Migrate {
    static final int BATCH = 1000;

    public static void main(String[] args) throws Exception {
        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");

        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))) {
            copyMap(hz.getMap("customers"), kv, "customer:", true);
            copyMap(hz.getMap("sessions"), kv, "session:", true);
            copyMap(hz.getMap("stock"), kv, "stock:", false);
            copyMap(hz.getMap("legacy"), kv, "order:", false);
            copyMultiMap(hz.getMultiMap("tags"), kv, "tags:");
            copyReplicatedMap(hz, kv);
            copyCollections(hz, kv);
            copyRingbuffer(hz.getRingbuffer("audit"), kv, "audit");
            // A counter's current value becomes a string that INCR/INCRBY work on.
            kv.set("counter:page-views", Long.toString(hz.getPNCounter("page-views").get()));
            System.out.println("pn counter page-views -> counter:page-views");
        }
        hz.shutdown();
    }

    // IMap -> one KiviDB key per entry. Objects become hashes, scalars become strings.
    // withTtl reads each entry's EntryView (one extra round trip per key) to carry its expiry.
    static void copyMap(IMap<Object, Object> map, JedisPooled kv, String prefix, boolean withTtl) {
        long n = 0, ttls = 0;
        Pipeline p = kv.pipelined();
        Iterator<Map.Entry<Object, Object>> it = map.iterator(BATCH);
        while (it.hasNext()) {
            Map.Entry<Object, Object> e = it.next();
            String key = prefix + e.getKey();
            Object v = e.getValue();
            if (v == null) continue;                       // expired or removed while we iterated
            p.del(key);
            if (v instanceof GenericRecord r) p.hset(key, toHash(r));
            else if (v instanceof LegacyOrder o) p.hset(key, Map.of("id", Long.toString(o.id), "sku", o.sku, "qty", Integer.toString(o.qty)));
            else if (v instanceof String || v instanceof Number || v instanceof Boolean) p.set(key, v.toString());
            else throw new IllegalStateException("No mapping for " + v.getClass() + " in " + map.getName());
            if (withTtl) {
                EntryView<Object, Object> ev = map.getEntryView(e.getKey());
                // An entry without TTL or max-idle reports Long.MAX_VALUE.
                if (ev != null && ev.getExpirationTime() != Long.MAX_VALUE) {
                    p.pexpireAt(key, ev.getExpirationTime());
                    ttls++;
                }
            }
            if (++n % BATCH == 0) p.sync();
        }
        p.sync();
        System.out.printf("map %s -> %s* : %d entries, %d with expiry%n", map.getName(), prefix, n, ttls);
    }

    // A Compact or Portable value read without its class. Every field becomes a hash field.
    static Map<String, String> toHash(GenericRecord r) {
        Map<String, String> h = new HashMap<>();
        for (String f : r.getFieldNames()) {
            Object v = switch (r.getFieldKind(f)) {
                case STRING -> r.getString(f);
                case BOOLEAN -> r.getBoolean(f);
                case INT8 -> r.getInt8(f);
                case INT16 -> r.getInt16(f);
                case INT32 -> r.getInt32(f);
                case INT64 -> r.getInt64(f);
                case FLOAT32 -> r.getFloat32(f);
                case FLOAT64 -> r.getFloat64(f);
                case NULLABLE_INT32 -> r.getNullableInt32(f);
                case NULLABLE_INT64 -> r.getNullableInt64(f);
                case NULLABLE_FLOAT64 -> r.getNullableFloat64(f);
                case NULLABLE_BOOLEAN -> r.getNullableBoolean(f);
                default -> throw new IllegalStateException("Add a mapping for field " + f + " of kind " + r.getFieldKind(f));
            };
            if (v != null) h.put(f, v.toString());   // a hash cannot hold null: leave the field out
        }
        return h;
    }

    // MultiMap -> one set per key (a MultiMap's values are a set by default).
    static void copyMultiMap(MultiMap<Object, Object> mm, JedisPooled kv, String prefix) {
        Pipeline p = kv.pipelined();
        long n = 0;
        for (Object k : mm.keySet()) {
            String key = prefix + k;
            p.del(key);
            p.sadd(key, mm.get(k).stream().map(Object::toString).toArray(String[]::new));
            if (++n % BATCH == 0) p.sync();
        }
        p.sync();
        System.out.printf("multimap %s -> %s* : %d keys%n", mm.getName(), prefix, n);
    }

    // ReplicatedMap -> one hash.
    static void copyReplicatedMap(HazelcastInstance hz, JedisPooled kv) {
        Map<String, String> h = new HashMap<>();
        hz.<Object, Object>getReplicatedMap("config").forEach((k, v) -> h.put(k.toString(), v.toString()));
        kv.del("config");
        if (!h.isEmpty()) kv.hset("config", h);
        System.out.printf("replicated map config -> config : %d fields%n", h.size());
    }

    // IQueue and IList -> lists (head first), ISet -> set. Iterating does not consume the queue.
    static void copyCollections(HazelcastInstance hz, JedisPooled kv) {
        kv.del("jobs", "events", "admins");
        Pipeline p = kv.pipelined();
        long q = 0, l = 0;
        for (Object o : hz.getQueue("jobs")) { p.rpush("jobs", o.toString()); q++; }
        for (Object o : hz.getList("events")) { p.rpush("events", o.toString()); l++; }
        String[] admins = hz.getSet("admins").stream().map(Object::toString).toArray(String[]::new);
        if (admins.length > 0) p.sadd("admins", admins);
        p.sync();
        System.out.printf("queue jobs -> jobs : %d, list events -> events : %d, set admins -> admins : %d%n", q, l, admins.length);
    }

    // Ringbuffer -> stream, oldest first. The stream assigns new IDs; the sequence is kept as a field.
    static void copyRingbuffer(Ringbuffer<Object> rb, JedisPooled kv, String key) throws Exception {
        kv.del(key);
        long seq = rb.headSequence(), tail = rb.tailSequence(), n = 0;
        Pipeline p = kv.pipelined();
        while (seq <= tail) {
            ReadResultSet<Object> rs = rb.readManyAsync(seq, 1, BATCH, null).toCompletableFuture().get();
            for (int i = 0; i < rs.size(); i++) {
                p.xadd(key, XAddParams.xAddParams(), Map.of("seq", Long.toString(rs.getSequence(i)), "value", rs.get(i).toString()));
                n++;
            }
            seq = rs.getNextSequenceToReadFrom();
            p.sync();
        }
        System.out.printf("ringbuffer %s -> stream %s : %d items%n", rb.getName(), key, n);
    }
}

Edit main to list your own structures and key prefixes, and copyMap to map your own value classes. Then run it:

export HZ_ADDRESS=hz-member-1.internal:5701 HZ_CLUSTER=prod
export KIVIDB_URL=redis://:$KIVIDB_PASSWORD@kividb.internal:6380
java -cp migrate.jar:hazelcast-5.7.0.jar:jedis-5.2.0.jar:commons-pool2-2.12.0.jar:slf4j-api-1.7.36.jar:gson-2.11.0.jar:json-20240303.jar Migrate

Notes on what it does:

• map.iterator(BATCH) pages through the map a partition at a time, so a large map is never held in the client's memory at once.
• Reading an entry's EntryView costs one round trip per key. Only pass withTtl = true for maps that use TTL or max-idle. It carries the absolute expiry time, so the key expires in KiviDB when it would have in Hazelcast.
• Reading entries counts as access for max-idle: the copy itself extends the idle deadline of every entry it reads.
• Iterating a queue reads it without taking items. Items taken while the copy runs are still copied; stop the consumers first, or see Cutover.
• Atomics and locks are left out on purpose: copy an IAtomicLong's value with kv.set(key, Long.toString(atomicLong.get())) at cutover, and don't copy locks (let them be taken again in KiviDB).

2. The same from Python

When values are strings, numbers or HazelcastJsonValue, the Hazelcast Python client (pip install hazelcast-python-client redis) is enough:

import os
import hazelcast
import redis

hz = hazelcast.HazelcastClient(
    cluster_name=os.environ.get("HZ_CLUSTER", "dev"),
    cluster_members=[os.environ.get("HZ_ADDRESS", "127.0.0.1:5701")],
)
kv = redis.Redis.from_url(os.environ.get("KIVIDB_URL", "redis://127.0.0.1:6380"))
BATCH = 1000


def copy_map(name, prefix, with_ttl):
    m = hz.get_map(name).blocking()
    keys = m.key_set()
    pipe = kv.pipeline(transaction=False)
    for i in range(0, len(keys), BATCH):
        chunk = keys[i:i + BATCH]
        values = m.get_all(chunk)
        for k, v in values.items():
            key = f"{prefix}{k}"
            pipe.set(key, str(v))
            if with_ttl:
                ev = m.get_entry_view(k)
                if ev is not None and ev.expiration_time != 2**63 - 1:
                    pipe.pexpireat(key, ev.expiration_time)
        pipe.execute()
    print(f"map {name}: {len(keys)} entries")


copy_map("sessions", "session:", with_ttl=True)
copy_map("stock", "stock:", with_ttl=False)

mm = hz.get_multi_map("tags").blocking()
pipe = kv.pipeline(transaction=False)
for k in mm.key_set():
    pipe.delete(f"tags:{k}")
    pipe.sadd(f"tags:{k}", *[str(v) for v in mm.get(k)])
pipe.execute()

kv.delete("jobs", "events", "admins", "config", "audit")
jobs = hz.get_queue("jobs").blocking().iterator()        # a snapshot: the queue is not consumed
if jobs:
    kv.rpush("jobs", *jobs)
events = hz.get_list("events").blocking().get_all()
if events:
    kv.rpush("events", *events)
admins = hz.get_set("admins").blocking().get_all()
if admins:
    kv.sadd("admins", *admins)
config = dict(hz.get_replicated_map("config").blocking().entry_set())
if config:
    kv.hset("config", mapping=config)

rb = hz.get_ringbuffer("audit").blocking()
seq, tail = rb.head_sequence(), rb.tail_sequence()
while seq <= tail:
    rs = rb.read_many(seq, 1, 500)                  # the Python client allows fewer than 1000 per read
    pipe = kv.pipeline(transaction=False)
    for i in range(rs.size):
        pipe.xadd("audit", {"seq": rs.get_sequence(i), "value": rs[i]})
    pipe.execute()
    seq = rs.next_sequence_to_read_from

kv.set("counter:page-views", hz.get_pn_counter("page-views").blocking().get())
print("done; KiviDB DBSIZE", kv.dbsize())
hz.shutdown()

key_set() brings every key of the map to the client at once; for maps with tens of millions of entries, use the Java program, which pages.

The Python client can't read Java-serialized values: get returns None for them, with no error. It reads Compact values only if you register a Python CompactSerializer for each type; without one, it raises No compact serializer is registered for the type name .... Use the Java program for those maps.

3. Follow changes while you copy

A copy of a live cluster is out of date as soon as it finishes. To keep KiviDB current until cutover, register an entry listener before the copy starts, and apply each event to KiviDB:

import com.hazelcast.client.HazelcastClient;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.EntryView;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.IMap;
import com.hazelcast.map.listener.*;
import com.hazelcast.nio.serialization.genericrecord.GenericRecord;
import redis.clients.jedis.JedisPooled;

// Replays an IMap's changes into KiviDB while the bulk copy runs and until cutover.
// Start it BEFORE Migrate, so nothing written during the copy is missed.
public class Follow {
    public static void main(String[] args) throws Exception {
        ClientConfig cc = new ClientConfig();
        cc.setClusterName(System.getenv().getOrDefault("HZ_CLUSTER", "dev"));
        cc.getNetworkConfig().addAddress(System.getenv().getOrDefault("HZ_ADDRESS", "127.0.0.1:5701"));
        HazelcastInstance hz = HazelcastClient.newHazelcastClient(cc);
        JedisPooled kv = new JedisPooled(java.net.URI.create(System.getenv().getOrDefault("KIVIDB_URL", "redis://127.0.0.1:6380")));
        for (String spec : args) {                         // e.g. customers=customer:
            String[] p = spec.split("=", 2);
            follow(hz.getMap(p[0]), kv, p[1]);
        }
        System.out.println("following; Ctrl-C at cutover");
        Thread.currentThread().join();
    }

    static void follow(IMap<Object, Object> map, JedisPooled kv, String prefix) {
        class L implements EntryAddedListener<Object, Object>, EntryUpdatedListener<Object, Object>,
                EntryRemovedListener<Object, Object>, EntryEvictedListener<Object, Object>,
                EntryExpiredListener<Object, Object> {
            void write(EntryEvent<Object, Object> e) {
                String key = prefix + e.getKey();
                Object v = e.getValue();
                if (v instanceof GenericRecord r) { kv.del(key); kv.hset(key, Migrate.toHash(r)); }
                else kv.set(key, String.valueOf(v));
                EntryView<Object, Object> ev = map.getEntryView(e.getKey());
                if (ev != null && ev.getExpirationTime() != Long.MAX_VALUE) kv.pexpireAt(key, ev.getExpirationTime());
            }
            public void entryAdded(EntryEvent<Object, Object> e) { write(e); }
            public void entryUpdated(EntryEvent<Object, Object> e) { write(e); }
            public void entryRemoved(EntryEvent<Object, Object> e) { kv.del(prefix + e.getKey()); }
            public void entryEvicted(EntryEvent<Object, Object> e) { kv.del(prefix + e.getKey()); }
            public void entryExpired(EntryEvent<Object, Object> e) { kv.del(prefix + e.getKey()); }
        }
        map.addEntryListener(new L(), true);               // true: events carry the value
    }
}

java -cp <same classpath> Follow customers=customer: sessions=session: &
java -cp <same classpath> Migrate

Entry listeners are not durable: if the follower's connection drops, the events in between are lost. If it restarts, run the copy again. There is also a small race: the copy can read an entry just before it changes and write the old value after the follower wrote the new one. Run Verify once writes are frozen, and copy again any key it reports. Only maps have entry listeners here; queues, lists and ringbuffers are copied once, at cutover.

4. Dual writes

Change the application to write every change to both Hazelcast and KiviDB (behind a setting), deploy that, run the copy once for the entries written before the deploy, then move reads to KiviDB. This needs no listener and survives restarts, but it is the most code to write and remove later.

5. From the system of record

If the maps are a cache in front of a database, loaded through a MapStore or MapLoader, KiviDB can be filled from that database instead: a job that reads the tables and writes the keys, or the application loading on a miss. Nothing is read from Hazelcast at all.

Hazelcast SQL

Hazelcast's SQL engine can read a map (SELECT * FROM customers) once a mapping exists, and is handy for checking counts. It needs the Jet engine enabled on the members (hazelcast.jet.enabled: true), which is off by default. The client program above doesn't need it.