ISSUE #1737: EntryMemTable.newEntry: always make a copy

Retaining a reference to that array assumes that the caller
won't reuse the array for something else -- an assumption
violated by Journal.scanJournal and probably other callers.

(bug W-5499346)
(rev cguttapalem)
Signed-off-by: Samuel Just <sjustsalesforce.com>

Author: 

Reviewers: Enrico Olivelli <eolivelligmail.com>, Matteo Merli <mmerliapache.org>, Sijie Guo <sijieapache.org>

This closes #1744 from athanatos/forupstream/wip-1737, closes #1737

(cherry picked from commit df65958)

Author: Sijie Guo <sijie@apache.org>
Author: Enrico Olivelli <eolivelli@apache.org>
Author: Matteo Merli <mmerli@apache.org>
Author: Samuel Just <sjust@salesforce.com>
Author: Nicolas Michael <nmichael@salesforce.com>
Author: Jia Zhai <zhaijia@apache.org>
Author: Sijie Guo <guosijie@gmail.com>
Author: cguttapalem <cguttapalem@salesforce.com>
Author: Ivan Kelly <ivank@apache.org>
Author: JV Jujjuri <vjujjuri@salesforce.com>
Author: Andrey Yegorov <dlg99@users.noreply.github.com>
Author: Ivan Kelly <ivan@ivankelly.net>
Author: Ethan Li <ethanopensource@gmail.com>
Author: arunlakshman <arunmuthuravi@gmail.com>
Author: Qi Wang <codingwangqi@gmail.com>
Author: Andrey Yegorov <ayegorov@salesforce.com>

Reviewers: Enrico Olivelli <eolivelli@gmail.com>, Sijie Guo <sijie@apache.org>

This closes #1746 from athanatos/forupstream/wip-1744-4.7, closes #1737

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryMemTable.java

@@ -352,13 +352,8 @@ private EntryKeyValue newEntry(long ledgerId, long entryId, final ByteBuffer ent
         int offset = 0;
         int length = entry.remaining();
 
-        if (entry.hasArray()) {
-            buf = entry.array();
-            offset = entry.arrayOffset();
-        } else {
-            buf = new byte[length];
-            entry.get(buf);
-        }
+        buf = new byte[length];
+        entry.get(buf);
         return new EntryKeyValue(ledgerId, entryId, buf, offset, length);
     }
 

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java

@@ -1543,6 +1543,18 @@ public int getSkipListArenaMaxAllocSize() {
         return getInt(SKIP_LIST_MAX_ALLOC_ENTRY, 128 * 1024);
     }
 
+    /**
+     * Set the max size we should allocate from the skiplist arena. Allocations
+     * larger than this should be allocated directly by the VM to avoid fragmentation.
+     *
+     * @param size max alloc size.
+     * @return server configuration object.
+     */
+    public ServerConfiguration setSkipListArenaMaxAllocSize(int size) {
+        setProperty(SKIP_LIST_MAX_ALLOC_ENTRY, size);
+        return this;
+    }
+
     /**
      * Should the data be fsynced on journal before acknowledgment.
      *

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieJournalTest.java

@@ -640,6 +640,39 @@ public void testTruncatedInEntryJournal() throws Exception {
         }
     }
 
+    /**
+     * Test journal replay with SortedLedgerStorage and a very small max
+     * arena size.
+     */
+    @Test
+    public void testSortedLedgerStorageReplayWithSmallMaxArenaSize() throws Exception {
+        File journalDir = createTempDir("bookie", "journal");
+        Bookie.checkDirectoryStructure(Bookie.getCurrentDirectory(journalDir));
+
+        File ledgerDir = createTempDir("bookie", "ledger");
+        Bookie.checkDirectoryStructure(Bookie.getCurrentDirectory(ledgerDir));
+
+        JournalChannel jc = writeV2Journal(
+                Bookie.getCurrentDirectory(journalDir), 100);
+
+        jc.fc.force(false);
+
+        writeIndexFileForLedger(Bookie.getCurrentDirectory(ledgerDir),
+                1, "testPasswd".getBytes());
+
+        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+        conf.setLedgerStorageClass("org.apache.bookkeeper.bookie.SortedLedgerStorage");
+        conf.setSkipListArenaMaxAllocSize(0);
+        conf.setJournalDirName(journalDir.getPath())
+                .setLedgerDirNames(new String[] { ledgerDir.getPath() });
+
+        Bookie b = new Bookie(conf);
+        b.readJournal();
+        b.ledgerStorage.flush();
+        b.readEntry(1, 80);
+        b.readEntry(1, 99);
+    }
+
     /**
      * Test partial index (truncate master key) with pre-v3 journals.
      */