remove() refactored as suggested on jbarrett's review

apache · Feb 4, 2022 · cbe859d · cbe859d
1 parent 43194d0
commit cbe859d
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 45 deletions.
diff --git a/...re/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java b/...re/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
@@ -284,10 +284,10 @@ public synchronized void remove() throws CacheException {
     if (peekedIds.isEmpty()) {
       return;
     }
-    boolean wasEmpty = lastDispatchedKey == lastDestroyedKey;
-    Long key = peekedIds.remove();
-    updateHeadKey(key);
-    lastDispatchedKey = key;
+    final boolean wasEmpty = lastDispatchedKey == lastDestroyedKey;
+    final Long key = peekedIds.remove();
+
+    preProcessRemovedKey(key);
 
     removeIndex(key);
     // Remove the entry at that key with a callback arg signifying it is
@@ -307,6 +307,8 @@ public synchronized void remove() throws CacheException {
       }
     }
 
+    postProcessRemovedKey();
+
     if (wasEmpty) {
       synchronized (this) {
         notifyAll();
@@ -320,6 +322,14 @@ public synchronized void remove() throws CacheException {
     }
   }
 
+  protected void preProcessRemovedKey(final Long key) {
+    updateHeadKey(key);
+    lastDispatchedKey = key;
+  }
+
+  protected void postProcessRemovedKey() {}
+
+
   /**
    * This method removes batchSize entries from the queue. It will only remove entries that were
    * previously peeked.

diff --git a/...apache/geode/cache/wan/internal/txgrouping/serial/TxGroupingSerialGatewaySenderQueue.java b/...apache/geode/cache/wan/internal/txgrouping/serial/TxGroupingSerialGatewaySenderQueue.java
@@ -27,9 +27,7 @@
 import java.util.function.Predicate;
 
 import org.apache.geode.annotations.VisibleForTesting;
-import org.apache.geode.cache.CacheException;
 import org.apache.geode.cache.CacheListener;
-import org.apache.geode.cache.EntryNotFoundException;
 import org.apache.geode.cache.TransactionId;
 import org.apache.geode.cache.asyncqueue.AsyncEvent;
 import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
@@ -198,37 +196,17 @@ protected List<KeyAndEventPair> getElementsMatching(Predicate<GatewaySenderEvent
   }
 
   @Override
-  public synchronized void remove() throws CacheException {
-    if (peekedIds.isEmpty()) {
-      return;
-    }
-    boolean wasEmpty = lastDispatchedKey == lastDestroyedKey;
-    Long key = peekedIds.remove();
+  protected void preProcessRemovedKey(Long key) {
     boolean isExtraPeekedId = extraPeekedIds.contains(key);
     if (!isExtraPeekedId) {
-      updateHeadKey(key);
-      lastDispatchedKey = key;
+      super.preProcessRemovedKey(key);
     } else {
       extraPeekedIdsRemovedButPreviousIdNotRemoved.add(key);
     }
-    removeIndex(key);
-    // Remove the entry at that key with a callback arg signifying it is
-    // a WAN queue so that AbstractRegionEntry.destroy can get the value
-    // even if it has been evicted to disk. In the normal case, the
-    // AbstractRegionEntry.destroy only gets the value in the VM.
-    try {
-      this.region.localDestroy(key, WAN_QUEUE_TOKEN);
-      this.stats.decQueueSize();
-    } catch (EntryNotFoundException ok) {
-      // this is acceptable because the conflation can remove entries
-      // out from underneath us.
-      if (logger.isDebugEnabled()) {
-        logger.debug(
-            "{}: Did not destroy entry at {} it was not there. It should have been removed by conflation.",
-            this, key);
-      }
-    }
+  }
 
+  @Override
+  protected void postProcessRemovedKey() {
     // For those extraPeekedIds removed that are consecutive to lastDispatchedKey:
     // - Update lastDispatchedKey with them so that they are removed
     // by the batch removal thread.
@@ -238,20 +216,7 @@ public synchronized void remove() throws CacheException {
     while (extraPeekedIdsRemovedButPreviousIdNotRemoved.contains(tmpKey = inc(tmpKey))) {
       extraPeekedIdsRemovedButPreviousIdNotRemoved.remove(tmpKey);
       extraPeekedIds.remove(tmpKey);
-      updateHeadKey(tmpKey);
-      lastDispatchedKey = tmpKey;
-    }
-
-    if (wasEmpty) {
-      synchronized (this) {
-        notifyAll();
-      }
-    }
-
-    if (logger.isDebugEnabled()) {
-      logger.debug(
-          "{}: Destroyed entry at key {} setting the lastDispatched Key to {}. The last destroyed entry was {}",
-          this, key, this.lastDispatchedKey, this.lastDestroyedKey);
+      super.preProcessRemovedKey(tmpKey);
     }
   }