Possible bug with MapStore

package com.hazelcast.bug;

import com.hazelcast.config.*;
import com.hazelcast.core.*;

import java.util.*;
import java.util.concurrent.*;


/**
 * Shutting down one instance leaves other instance with fewer keys.
 */
public class MapStoreMigrationBug implements Runnable {

    public static void main(String... args) {
        try {
            new MapStoreMigrationBug().run();
        } finally {
            Hazelcast.shutdownAll();
        }
    }

    public void run() {
        HazelcastInstance hc1 = Hazelcast.newHazelcastInstance(getConfig("hc1"));
        HazelcastInstance hc2 = Hazelcast.newHazelcastInstance(getConfig("hc2"));

        IMap<Integer, String> m1 = hc1.getMap(MAP_NAME);
        IMap<Integer, String> m2 = hc2.getMap(MAP_NAME);

        System.out.println(
            "***** Verify that local key sets are partitioned as expected. *****");
        System.out.printf("hc1.testMap.localKeySet() = %s%n", asString(m1.localKeySet()));
        System.out.printf("hc2.testMap.localKeySet() = %s%n", asString(m2.localKeySet()));

        System.out.println(
            "***** Verify that hc2 sees \"changed one\". *****");
        m1.put(1, "changed one");
        String changedOne = m2.get(1);
        System.out.printf("new value for 1 put in hc1, read in hc2 as: %s%n", changedOne);

        System.out.println(
            "***** Verify that the (non-local) key sets have all keys. *****");
        System.out.printf("hc1.testMap.keySet() = %s%n", asString(m1.keySet()));
        System.out.printf("hc2.testMap.keySet() = %s%n", asString(m2.keySet()));

        hc1.getLifecycleService().shutdown();

        try {
            TimeUnit.SECONDS.sleep(15);
            System.out.println(
                "***** BUG? hc2 keySet has only local plus changed one. *****");
            System.out.printf(
                "after hc1 shutdown, hc2.testMap.keySet() = %s%n",
                asString(m2.keySet()));
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            return;
        }
    }

    private String asString(Set<?> set) {
        StringBuilder buf = new StringBuilder("{ ");
        for (Object element : set) {
            buf.append(element.toString()).append(", ");
        }
        buf.append("}");
        return buf.toString();
    }


    static class Factory implements MapStoreFactory {
        Factory(String factoryName) {
            this.factoryName = factoryName;
        }

        private final String factoryName;

        @Override public MapLoader<?, ?> newMapStore(String mapName, Properties properties) {
            return createMapStore(mapName, properties);
        }

        private MapLoader<Integer, String> createMapStore(
                final String mapName,
                Properties properties) {

            return new MapStore<Integer, String>() {
                @Override public String load(Integer key) {
                    String value = STORE.get(key);
                    System.out.printf(
                        "%s.%s.load(\"%d\")=%s%n",
                        factoryName, mapName, key, value);
                    return value;
                }

                @Override public Map<Integer, String> loadAll(Collection<Integer> keys) {
                    Map<Integer, String> result = new HashMap<Integer, String>();
                    for (Integer key : keys) {
                        String value = load(key);
                        if (value != null) {
                            result.put(key, value);
                        }
                    }
                    return result;
                }

                @Override public Set<Integer> loadAllKeys() {
                    return STORE.keySet();
                }

                @Override public void store(Integer key, String value) {
                    System.out.printf(
                        "%s.%s.store(\"%d\", \"%s\")%n",
                        factoryName, mapName, key, value);
                    STORE.put(key, value);
                }

                @Override public void storeAll(Map<Integer, String> map) {
                    for (Map.Entry<Integer, String> entry : map.entrySet()) {
                        store(entry.getKey(), entry.getValue());
                    }
                }

                @Override public void delete(Integer key) {
                    System.out.printf(
                        "%s.%s.delete(\"%d\")%n",
                        factoryName, mapName, key);
                    STORE.remove(key);
                }

                @Override public void deleteAll(Collection<Integer> keys) {
                    for (Integer key : STORE.keySet()) {
                        delete(key);
                    }
                }
            };
        }
    }

    private static Config getConfig(String factoryName) {
        Config config = new Config();
        config
            .getMapConfig(MAP_NAME)
            .setMapStoreConfig(new MapStoreConfig()
                .setWriteDelaySeconds(1)
                .setFactoryImplementation(new Factory(factoryName)));
        return config;
    }


    private static final String MAP_NAME = "testMap";

    private static final ConcurrentMap<Integer, String> STORE =
        new ConcurrentHashMap<Integer, String>();

    static {
        STORE.put(1, "one");
        STORE.put(102, "two");
        STORE.put(3, "three");
        STORE.put(104, "four");
        STORE.put(5, "five");
        STORE.put(106, "six");
        STORE.put(7, "seven");
        STORE.put(108, "eight");
        STORE.put(9, "nine");
        STORE.put(110, "ten");
        STORE.put(11, "eleven");
        STORE.put(112, "twelve");
    }
}