Queue Memory Leak Test 3.0

QueueMemoryLeak.java

public class QueueMemoryLeak {

    static {
        System.setProperty(GroupProperties.PROP_WAIT_SECONDS_BEFORE_JOIN, "0");
        System.setProperty("java.net.preferIPv4Stack", "true");
        System.setProperty("hazelcast.local.localAddress", "127.0.0.1");
        System.setProperty("hazelcast.version.check.enabled", "false");
        System.setProperty("hazelcast.socket.bind.any", "false");

        Random rand = new Random();
        int g1 = rand.nextInt(255);
        int g2 = rand.nextInt(255);
        int g3 = rand.nextInt(255);
        System.setProperty("hazelcast.multicast.group", "224." + g1 + "." + g2 + "." + g3);

    }

    public static void main(String[] args) throws InterruptedException {

        final CountDownLatch latch = new CountDownLatch(4);

        new Thread(){
            public void run() {
                final HazelcastInstance instance = Hazelcast.newHazelcastInstance();
                final IQueue<TestItem> q = instance.getQueue("q");

                for (int i=0; i<600*1000; i++) {
                    q.offer(new TestItem("item-"+i, i));
                }
                latch.countDown();
                System.err.println("offer done");
                

            }
        }.start();


        Thread.sleep(8000);


        System.err.println("starting consume");

        for (int i=0; i<3; i++){
            final int id = i;

            new Thread(){
                public void run() {
                    final HazelcastInstance instance = Hazelcast.newHazelcastInstance();
                    final IQueue<TestItem> q = instance.getQueue("q");

                    try {
                        while (true) {
                            final TestItem removed = q.remove();
                            if (removed.type % 1000 == 0) {
                                System.err.println("removed: " + removed.name);
                            }
                            Thread.sleep(1);
                        }
                    } finally {
                        System.err.println("thread["+id+"] done");
                        latch.countDown();
                    }

                }
            }.start();
        }


        latch.await();

        System.err.println("all done!");

    }

}

public class TestItem implements Serializable {

    String name;

    int type;

    public TestItem() {
    }

    public TestItem(String name, int type) {
        this.name = name;
        this.type = type;
    }



}