Using KeyValueStorage.Batch more rigorously (#4576)

Every time use KeyValueStorage.Batch, close it in finally

Author: StevenLuMT

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndex.java

@@ -147,10 +147,10 @@ private long getLastEntryInLedgerInternal(long ledgerId) throws IOException {
     }
 
     public void addLocation(long ledgerId, long entryId, long location) throws IOException {
-        Batch batch = locationsDb.newBatch();
-        addLocation(batch, ledgerId, entryId, location);
-        batch.flush();
-        batch.close();
+        try (Batch batch = locationsDb.newBatch()) {
+            addLocation(batch, ledgerId, entryId, location);
+            batch.flush();
+        }
     }
 
     public Batch newBatch() {
@@ -178,18 +178,17 @@ public void updateLocations(Iterable<EntryLocation> newLocations) throws IOExcep
             log.debug("Update locations -- {}", Iterables.size(newLocations));
         }
 
-        Batch batch = newBatch();
-        // Update all the ledger index pages with the new locations
-        for (EntryLocation e : newLocations) {
-            if (log.isDebugEnabled()) {
-                log.debug("Update location - ledger: {} -- entry: {}", e.ledger, e.entry);
-            }
+        try (Batch batch = newBatch()) {
+            // Update all the ledger index pages with the new locations
+            for (EntryLocation e : newLocations) {
+                if (log.isDebugEnabled()) {
+                    log.debug("Update location - ledger: {} -- entry: {}", e.ledger, e.entry);
+                }
 
-            addLocation(batch, e.ledger, e.entry, e.location);
+                addLocation(batch, e.ledger, e.entry, e.location);
+            }
+            batch.flush();
         }
-
-        batch.flush();
-        batch.close();
     }
 
     public void delete(long ledgerId) throws IOException {