Merge pull request #1567 from cshannon/AMQ-9824

AMQ-9824 - Cleanup code in KahaDB classes

Author: cshannon

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java

@@ -442,31 +442,28 @@ private void corruptBatchEndEof(int id) throws Exception{
     private void corruptOrderIndex(final int num, final int size) throws Exception {
         //This is because of AMQ-6097, now that the MessageOrderIndex stores the size in the Location,
         //we need to corrupt that value as well
-        final KahaDBStore kahaDbStore = (KahaDBStore) ((KahaDBPersistenceAdapter) broker.getPersistenceAdapter()).getStore();
-        kahaDbStore.indexLock.writeLock().lock();
+        final KahaDBStore kahaDbStore = ((KahaDBPersistenceAdapter) broker.getPersistenceAdapter()).getStore();
+        kahaDbStore.indexLock.lock();
         try {
-            kahaDbStore.pageFile.tx().execute(new Transaction.Closure<IOException>() {
-                @Override
-                public void execute(Transaction tx) throws IOException {
-                    StoredDestination sd = kahaDbStore.getStoredDestination(kahaDbStore.convert(
-                        (ActiveMQQueue)destination), tx);
-                    int i = 1;
-                    for (Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx); iterator.hasNext();) {
-                        Entry<Long, MessageKeys> entry = iterator.next();
-                        if (i == num) {
-                            //change the size value to the wrong size
-                            sd.orderIndex.get(tx, entry.getKey());
-                            MessageKeys messageKeys = entry.getValue();
-                            messageKeys.location.setSize(size);
-                            sd.orderIndex.put(tx, sd.orderIndex.lastGetPriority(), entry.getKey(), messageKeys);
-                            break;
-                        }
-                        i++;
+            kahaDbStore.pageFile.tx().execute(tx -> {
+                StoredDestination sd = kahaDbStore.getStoredDestination(kahaDbStore.convert(
+                    (ActiveMQQueue)destination), tx);
+                int i = 1;
+                for (Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx); iterator.hasNext();) {
+                    Entry<Long, MessageKeys> entry = iterator.next();
+                    if (i == num) {
+                        //change the size value to the wrong size
+                        sd.orderIndex.get(tx, entry.getKey());
+                        MessageKeys messageKeys = entry.getValue();
+                        messageKeys.location.setSize(size);
+                        sd.orderIndex.put(tx, sd.orderIndex.lastGetPriority(), entry.getKey(), messageKeys);
+                        break;
                     }
+                    i++;
                 }
             });
         } finally {
-            kahaDbStore.indexLock.writeLock().unlock();
+            kahaDbStore.indexLock.unlock();
         }
     }
 

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java

@@ -164,22 +164,19 @@ private void corruptIndex() throws IOException {
 
         //blow up the index
         try {
-            store.indexLock.writeLock().lock();
-            pageFile.tx().execute(new Transaction.Closure<IOException>() {
-                @Override
-                public void execute(Transaction tx) throws IOException {
-                    for (Iterator<Entry<String, StoredDestination>> iterator = metadata.destinations.iterator(tx); iterator
-                            .hasNext();) {
-                        Entry<String, StoredDestination> entry = iterator.next();
-                        entry.getValue().orderIndex.nextMessageId = -100;
-                        entry.getValue().orderIndex.defaultPriorityIndex.clear(tx);
-                        entry.getValue().orderIndex.lowPriorityIndex.clear(tx);
-                        entry.getValue().orderIndex.highPriorityIndex.clear(tx);
-                    }
+            store.indexLock.lock();
+            pageFile.tx().execute(tx -> {
+                for (Iterator<Entry<String, StoredDestination>> iterator = metadata.destinations.iterator(tx); iterator
+                        .hasNext();) {
+                    Entry<String, StoredDestination> entry = iterator.next();
+                    entry.getValue().orderIndex.nextMessageId = -100;
+                    entry.getValue().orderIndex.defaultPriorityIndex.clear(tx);
+                    entry.getValue().orderIndex.lowPriorityIndex.clear(tx);
+                    entry.getValue().orderIndex.highPriorityIndex.clear(tx);
                 }
             });
         } finally {
-            store.indexLock.writeLock().unlock();
+            store.indexLock.unlock();
         }
     }
 

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/JournalCorruptionEofIndexRecoveryTest.java

@@ -62,11 +62,11 @@ public void forceCleanup() throws IOException {
 
         public int getFileMapSize() throws IOException {
             // ensure save memory publishing, use the right lock
-            indexLock.readLock().lock();
+            indexLock.lock();
             try {
                 return getJournal().getFileMap().size();
             } finally {
-                indexLock.readLock().unlock();
+                indexLock.unlock();
             }
         }
     }

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/KahaDBStoreOpenWireVersionTest.java

@@ -65,11 +65,11 @@ public void forceCleanup() throws IOException {
 
         public int getFileMapSize() throws IOException {
             // ensure save memory publishing, use the right lock
-            indexLock.readLock().lock();
+            indexLock.lock();
             try {
                 return getJournal().getFileMap().size();
             } finally {
-                indexLock.readLock().unlock();
+                indexLock.unlock();
             }
         }
     }

activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2982Test.java

@@ -61,7 +61,7 @@ public void testLocationIndexMatchesOrderIndex() throws Exception {
 
         //Iterate over the order index and add up the size of the messages to compare
         //to the location index
-        kahaDbStore.indexLock.readLock().lock();
+        kahaDbStore.indexLock.lock();
         try {
             long size = kahaDbStore.pageFile.tx().execute(new Transaction.CallableClosure<Long, IOException>() {
                 @Override
@@ -79,7 +79,7 @@ public Long execute(Transaction tx) throws IOException {
             assertEquals("Order index size values don't match message size",
                     size, messageStore.getMessageSize());
         } finally {
-            kahaDbStore.indexLock.readLock().unlock();
+            kahaDbStore.indexLock.unlock();
         }
     }