queue: Allow stuck entries (same node) to be reaped. See #169

queue/src/main/java/org/killbill/queue/DBBackedQueue.java

@@ -282,39 +282,27 @@ public Void inTransaction(final QueueSqlDao<T> transactional, final TransactionS
                     return null;
                 }
 
-                final Collection<T> entriesToMove = new ArrayList<T>(entriesLeftBehind.size());
                 final List<T> entriesToReInsert = new ArrayList<T>(entriesLeftBehind.size());
-                final List<T> stuckEntries = new LinkedList<T>();
                 final List<T> lateEntries = new LinkedList<T>();
                 for (final T entryLeftBehind : entriesLeftBehind) {
                     // entryIsBeingProcessedByThisNode is a sign of a stuck entry on this node
-                    final boolean entryIsBeingProcessedByThisNode = owner.equals(entryLeftBehind.getProcessingOwner());
                     // entryCreatedByThisNodeAndNeverProcessed is likely a sign of the queue being late
                     final boolean entryCreatedByThisNodeAndNeverProcessed = owner.equals(entryLeftBehind.getCreatingOwner()) && entryLeftBehind.getProcessingOwner() == null;
-                    if (entryIsBeingProcessedByThisNode) {
-                        // See https://github.com/killbill/killbill-commons/issues/47
-                        stuckEntries.add(entryLeftBehind);
-                    } else if (entryCreatedByThisNodeAndNeverProcessed) {
+                    if (entryCreatedByThisNodeAndNeverProcessed) {
                         lateEntries.add(entryLeftBehind);
-                    } else {
-                        // Fields will be reset appropriately in insertReapedEntriesFromTransaction
-                        entriesToReInsert.add(entryLeftBehind);
-
+                    } else { /* This includes entryIsBeingProcessedByThisNode (owner.equals(entryLeftBehind.getProcessingOwner())). See https://github.com/killbill/killbill-commons/issues/169 */
                         // Set the status to REAPED in the history table
                         entryLeftBehind.setProcessingState(PersistentQueueEntryLifecycleState.REAPED);
-                        entriesToMove.add(entryLeftBehind);
+                        entriesToReInsert.add(entryLeftBehind);
                     }
                 }
 
-                if (!stuckEntries.isEmpty()) {
-                    log.warn("{} reapEntries: stuck queue entries {}", DB_QUEUE_LOG_ID, stuckEntries);
-                }
                 if (!lateEntries.isEmpty()) {
                     log.warn("{} reapEntries: late queue entries {}", DB_QUEUE_LOG_ID, lateEntries);
                 }
 
                 if (!entriesToReInsert.isEmpty()) {
-                    moveEntriesToHistoryFromTransaction(transactional, entriesToMove);
+                    moveEntriesToHistoryFromTransaction(transactional, entriesToReInsert);
                     insertReapedEntriesFromTransaction(transactional, entriesToReInsert, now);
                     log.warn("{} reapEntries: {} entries were reaped by {} {}",
                              DB_QUEUE_LOG_ID,

queue/src/test/java/org/killbill/TestSetup.java

@@ -53,6 +53,9 @@ public class TestSetup {
 
     private static final String TEST_DB_PROPERTY_PREFIX = "org.killbill.billing.dbi.test.";
 
+    protected static final Long SEARCH_KEY_1 = 1L;
+    protected static final Long SEARCH_KEY_2 = 2L;
+
     protected EmbeddedDB embeddedDB;
 
     protected DBI dbi;

queue/src/test/java/org/killbill/queue/TestReaper.java

@@ -42,8 +42,6 @@
 
 public class TestReaper extends TestSetup {
 
-    private static final Long SEARCH_KEY_1 = 1L;
-    private static final Long SEARCH_KEY_2 = 2L;
 
     private DBBackedQueue<BusEventModelDao> queue;
     private PersistentBusSqlDao sqlDao;
@@ -102,26 +100,28 @@ public void testReapEntries() {
         queue.reapEntries(now.minus(config.getReapThreshold().getMillis()).toDate());
 
         final List<BusEventModelDao> readyEntriesAfterReaping = sqlDao.getReadyEntries(now.toDate(), 10, CreatorName.get(), config.getTableName());
-        assertEquals(readyEntriesAfterReaping.size(), 2);
+        assertEquals(readyEntriesAfterReaping.size(), 3);
         assertEquals(readyEntriesAfterReaping.get(0).getRecordId(), (Long) 1L);
         assertTrue(readyEntriesAfterReaping.get(1).getRecordId() > (Long) 6L);
+        assertTrue(readyEntriesAfterReaping.get(2).getRecordId() > (Long) 6L);
 
         final List<BusEventModelDao> readyOrInProcessingAfterReaping = Iterators.toUnmodifiableList(sqlDao.getReadyOrInProcessingQueueEntriesForSearchKeys(SEARCH_KEY_1, SEARCH_KEY_2, config.getTableName()));
         assertEquals(readyOrInProcessingAfterReaping.size(), 6);
         assertEquals(readyOrInProcessingAfterReaping.get(0).getRecordId(), (Long) 1L);
         assertEquals(readyOrInProcessingAfterReaping.get(1).getRecordId(), (Long) 2L);
         assertEquals(readyOrInProcessingAfterReaping.get(2).getRecordId(), (Long) 3L);
         assertEquals(readyOrInProcessingAfterReaping.get(3).getRecordId(), (Long) 4L);
-        // That stuck entry hasn't moved (https://github.com/killbill/killbill-commons/issues/47)
-        assertEquals(readyOrInProcessingAfterReaping.get(4).getRecordId(), (Long) 5L);
-        // New (reaped) one
+        // New (reaped) ones
+        assertTrue(readyOrInProcessingAfterReaping.get(4).getRecordId() > (Long) 6L);
         assertTrue(readyOrInProcessingAfterReaping.get(5).getRecordId() > (Long) 6L);
 
         // Check history table
         final List<BusEventModelDao> historicalQueueEntries = Iterators.toUnmodifiableList(sqlDao.getHistoricalQueueEntriesForSearchKeys(SEARCH_KEY_1, SEARCH_KEY_2, config.getHistoryTableName()));
-        assertEquals(historicalQueueEntries.size(), 1);
+        assertEquals(historicalQueueEntries.size(), 2);
         assertEquals(historicalQueueEntries.get(0).getProcessingState(), PersistentQueueEntryLifecycleState.REAPED);
-        assertEquals(historicalQueueEntries.get(0).getUserToken(), readyOrInProcessingAfterReaping.get(5).getUserToken());
+        assertEquals(historicalQueueEntries.get(0).getUserToken(), readyOrInProcessingAfterReaping.get(4).getUserToken());
+        assertEquals(historicalQueueEntries.get(1).getProcessingState(), PersistentQueueEntryLifecycleState.REAPED);
+        assertEquals(historicalQueueEntries.get(1).getUserToken(), readyOrInProcessingAfterReaping.get(5).getUserToken());
     }
 
     private BusEventModelDao createEntry(final long recordId,

queue/src/test/java/org/killbill/queue/TestReaperIntegration.java

@@ -31,6 +31,7 @@
 import org.killbill.TestSetup;
 import org.killbill.bus.DefaultPersistentBus;
 import org.killbill.bus.api.BusEvent;
+import org.killbill.bus.api.BusEventWithMetadata;
 import org.killbill.bus.api.PersistentBus.EventBusException;
 import org.killbill.bus.api.PersistentBusConfig;
 import org.killbill.bus.dao.BusEventModelDao;
@@ -39,6 +40,7 @@
 import org.killbill.commons.eventbus.Subscribe;
 import org.killbill.queue.api.PersistentQueueEntryLifecycleState;
 import org.skife.config.TimeSpan;
+import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
@@ -164,9 +166,17 @@ public void testWithStuckEntryProcessedByThisNode() throws EventBusException, Js
 
         // Trigger reaper
         clock.addDeltaFromReality(config.getReapThreshold().getMillis());
-        // It's a no-op though (it won't reap itself)
-        handler.ensureNotSeen(event2);
-        handler.assertSeenEvents(2);
+        // Give a chance to the reaper to run
+        Thread.sleep(500);
+
+        clock.addDeltaFromReality(1000);
+        handler.waitFor(event2);
+
+        // See https://github.com/killbill/killbill-commons/issues/169
+        handler.assertSeenEvents(3);
+        final Iterable<BusEventWithMetadata<BusEvent>> result = bus.getHistoricalBusEventsForSearchKey2(now, SEARCH_KEY_2);
+        final long nbItems = result.spliterator().getExactSizeIfKnown();
+        assertEquals(nbItems, 4);
     }
 
 
@@ -278,8 +288,8 @@ public DummyEvent(@JsonProperty("name") final String name,
 
         public DummyEvent() {
             this(UUID.randomUUID().toString(),
-                 System.currentTimeMillis(),
-                 System.currentTimeMillis(),
+                 SEARCH_KEY_1,
+                 SEARCH_KEY_2,
                  UUID.randomUUID());
         }