### Question:
EntryMemTable implements AutoCloseable { long flushSnapshot(final SkipListFlusher flusher, Checkpoint checkpoint) throws IOException { long size = 0; if (this.snapshot.compareTo(checkpoint) < 0) { long ledger, ledgerGC = -1; synchronized (this) { EntrySkipList keyValues = this.snapshot; if (keyValues.compareTo(checkpoint) < 0) { for (EntryKey key : keyValues.keySet()) { EntryKeyValue kv = (EntryKeyValue) key; size += kv.getLength(); ledger = kv.getLedgerId(); if (ledgerGC != ledger) { try { flusher.process(ledger, kv.getEntryId(), kv.getValueAsByteBuffer()); } catch (NoLedgerException exception) { ledgerGC = ledger; } } } memTableStats.getFlushBytesCounter().add(size); clearSnapshot(keyValues); } } } skipListSemaphore.release((int) size); return size; } EntryMemTable(final ServerConfiguration conf, final CheckpointSource source,
                         final StatsLogger statsLogger); long flush(SkipListFlusher flusher, Checkpoint checkpoint); long addEntry(long ledgerId, long entryId, final ByteBuffer entry, final CacheCallback cb); EntryKeyValue getEntry(long ledgerId, long entryId); EntryKeyValue getLastEntry(long ledgerId); @Override void close();  }### Answer:
@Test public void testFlushSnapshot() throws IOException { HashSet<EntryKeyValue> keyValues = new HashSet<EntryKeyValue>(); Set<EntryKeyValue> flushedKVs = Collections.newSetFromMap(new ConcurrentHashMap<EntryKeyValue, Boolean>()); KVFLusher flusher = new KVFLusher(flushedKVs); byte[] data = new byte[10]; for (long entryId = 1; entryId < 100; entryId++) { for (long ledgerId = 1; ledgerId < 100; ledgerId++) { random.nextBytes(data); assertTrue(ledgerId + ":" + entryId + " is duplicate in mem-table!", memTable.addEntry(ledgerId, entryId, ByteBuffer.wrap(data), this) != 0); assertTrue(ledgerId + ":" + entryId + " is duplicate in hash-set!", keyValues.add(memTable.getEntry(ledgerId, entryId))); if (random.nextInt(16) == 0) { if (null != memTable.snapshot()) { if (random.nextInt(2) == 0) { memTable.flush(flusher); } } } } } memTable.flush(flusher, Checkpoint.MAX); for (EntryKeyValue kv : keyValues) { assertTrue("kv " + kv.toString() + " was not flushed!", flushedKVs.contains(kv)); } }