add test code.

pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java

@@ -178,6 +178,12 @@ public void testExpandAndShrink() {
         assertEquals(map.capacity(), 8);
     }
 
+    /**
+     * Test concurrent write and iterate, and verify that
+     * if i can see the pair (n,n,n,n) and n>0, i can see the pair (n-1,n-1,n-1,n-1)
+     * because the pair (n,n,n,n) is inserted after the pair (n-1,n-1,n-1,n-1)
+     * @throws Throwable
+     */
     @Test
     public void testConcurrentWriteAndIterate()  throws Throwable {
         ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder()
@@ -244,6 +250,89 @@ public void testConcurrentWriteAndIterate()  throws Throwable {
         executor.shutdown();
     }
 
+    /**
+     * Test concurrent write and iterate, and verify that
+     * In the results of iteration of map, there can't be two mapping pair
+     * with the same key whatever their value is same or not.
+     * @throws Throwable
+     */
+    @Test
+    public void testConcurrentWriteAndIterateV2()  throws Throwable {
+        ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder()
+                .expectedItems(2)
+                .concurrencyLevel(1)
+                .autoShrink(true)
+                .mapIdleFactor(0.25f)
+                .build();
+        assertEquals(map.capacity(), 4);
+
+        ExecutorService executor = Executors.newCachedThreadPool();
+        final int readThreads = 16;
+        final int writeThreads = 1;
+        final int n = 2_000;
+        CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads);
+        Future<?> future = null;
+        AtomicReference<Exception> ex = new AtomicReference<>();
+
+        for (int i = 0; i < readThreads; i++) {
+            executor.submit(() -> {
+                try {
+                    barrier.await();
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+                Map<Long, Long> keys = new HashMap<>();
+                while (true) {
+                    map.forEach((k1, k2, v1, v2) -> {
+                        if(keys.containsKey(k1)) {
+                            ex.set(new Exception("Duplicate mapping pair with same key: (" + k1
+                                    + "," + keys.get(k1) + "), (" + k1 + "," + v1 + ")"));
+                        }
+                        keys.put(k1, v1);
+                    });
+                    keys.clear();
+                }
+            });
+        }
+
+        future = executor.submit(() -> {
+            try {
+                barrier.await();
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+
+            for (int i = 0; i < n; i++) {
+                // expand hashmap
+                assertTrue(map.put(9, 9, 99, 99));
+                assertTrue(map.put(10, 10, 1010, 1010));
+
+                // shrink hashmap
+                map.clear();
+                assertEquals(map.capacity(), 4);
+
+                // with different value
+                assertTrue(map.put(10, 10, 99, 99));
+                assertTrue(map.put(9, 9, 1010, 1010));
+
+//                // with same value
+//                assertTrue(map.put(10, 10, 1010, 1010));
+//                assertTrue(map.put(9, 9, 99, 99));
+
+                // shrink hashmap
+                map.clear();
+                assertEquals(map.capacity(), 4);
+            }
+        });
+
+        future.get();
+        System.out.println(ex.get());
+        assertTrue(ex.get() == null);
+        // shut down pool
+        executor.shutdown();
+    }
+
+
 
     @Test
     public void testExpandShrinkAndClear() {