GEODE-2485: fix leak in tx suspend/resume

test now closes the cache it creates

geode-core/src/main/java/org/apache/geode/cache/query/internal/DefaultQuery.java

@@ -599,7 +599,7 @@ public Object executeUsingContext(ExecutionContext context) throws FunctionDomai
       updateStatistics(endTime - startTime);
       pdxClassToFieldsMap.remove();
       pdxClassToMethodsMap.remove();
-      ((TXManagerImpl) this.cache.getCacheTransactionManager()).resume(tx);
+      ((TXManagerImpl) this.cache.getCacheTransactionManager()).internalResume(tx);
     }
   }
 

geode-core/src/main/java/org/apache/geode/cache/query/internal/index/IndexManager.java

@@ -407,7 +407,7 @@ public Index createIndex(String indexName, IndexType indexType, String origIndex
       DefaultQuery.setPdxReadSerialized(this.region.getCache(), oldReadSerialized);
 
       if (tx != null) {
-        ((TXManagerImpl) this.region.getCache().getCacheTransactionManager()).resume(tx);
+        ((TXManagerImpl) this.region.getCache().getCacheTransactionManager()).internalResume(tx);
       }
     }
   }
@@ -1138,7 +1138,7 @@ private void processAction(RegionEntry entry, int action, int opCode) throws Que
     } finally {
       DefaultQuery.setPdxReadSerialized(this.region.getCache(), false);
       if (tx != null) {
-        ((TXManagerImpl) this.region.getCache().getCacheTransactionManager()).resume(tx);
+        ((TXManagerImpl) this.region.getCache().getCacheTransactionManager()).internalResume(tx);
       }
       getCachePerfStats().endIndexUpdate(startPA);
     }

geode-core/src/main/java/org/apache/geode/internal/SystemTimer.java

@@ -43,7 +43,7 @@
  *      TODO -- with Java 1.5, this will be a template type so that the swarm's class can be
  *      specified.
  */
-public final class SystemTimer {
+public class SystemTimer {
   private static final Logger logger = LogService.getLogger();
 
   final static private boolean isIBM =

geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java

@@ -3949,7 +3949,7 @@ private void setMemoryThresholdReachedCounterTrue(final DistributedMember idm) {
   protected VersionTag fetchRemoteVersionTag(Object key) {
     VersionTag tag = null;
     assert this.dataPolicy != DataPolicy.REPLICATE;
-    TransactionId txId = cache.getCacheTransactionManager().suspend();
+    final TXStateProxy tx = cache.getTXMgr().internalSuspend();
     try {
       boolean retry = true;
       InternalDistributedMember member = getRandomReplicate();
@@ -3971,8 +3971,8 @@ protected VersionTag fetchRemoteVersionTag(Object key) {
         }
       }
     } finally {
-      if (txId != null) {
-        cache.getCacheTransactionManager().resume(txId);
+      if (tx != null) {
+        cache.getTXMgr().internalResume(tx);
       }
     }
     return tag;

geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java

@@ -2437,7 +2437,7 @@ public void close(String reason, Throwable systemFailureCause, boolean keepalive
         }
         ((DynamicRegionFactoryImpl) DynamicRegionFactory.get()).close();
         if (this.txMgr != null) {
-          this.txMgr.resume(tx);
+          this.txMgr.internalResume(tx);
         }
         TXCommitMessage.getTracker().clearForCacheClose();
       }
@@ -3936,6 +3936,32 @@ public SystemTimer getCCPTimer() {
     }
   }
 
+  /**
+   * For use by unit tests to inject a mocked ccpTimer
+   */
+  void setCCPTimer(SystemTimer ccpTimer) {
+    this.ccpTimer = ccpTimer;
+  }
+
+  static final int PURGE_INTERVAL = 1000;
+  private int cancelCount = 0;
+
+  /**
+   * Does a periodic purge of the CCPTimer to prevent a large number of cancelled tasks from
+   * building up in it. See GEODE-2485.
+   */
+  public void purgeCCPTimer() {
+    synchronized (ccpTimerMutex) {
+      if (ccpTimer != null) {
+        cancelCount++;
+        if (cancelCount == PURGE_INTERVAL) {
+          cancelCount = 0;
+          ccpTimer.timerPurge();
+        }
+      }
+    }
+  }
+
   /**
    * @see LocalRegion
    */

geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java

@@ -6556,7 +6556,7 @@ void basicDestroyRegion(RegionEventImpl event, boolean cacheWrite, boolean lock,
         }
       }
     } finally {
-      this.cache.getTXMgr().resume(tx);
+      this.cache.getTXMgr().internalResume(tx);
     }
   }
 
@@ -7180,7 +7180,7 @@ void basicInvalidateRegion(RegionEventImpl event) {
         dispatchListenerEvent(EnumListenerEvent.AFTER_REGION_INVALIDATE, event);
       }
     } finally {
-      this.cache.getTXMgr().resume(tx);
+      this.cache.getTXMgr().internalResume(tx);
     }
   }
 
@@ -9572,7 +9572,7 @@ Map basicGetAll(Collection keys, Object callback) {
                       key);
                 }
               } finally {
-                this.cache.getTXMgr().resume(tx);
+                this.cache.getTXMgr().internalResume(tx);
               }
               getCachePerfStats().endPut(startPut, event.isOriginRemote());
             }

geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataView.java

@@ -58,7 +58,7 @@ public Entry getEntry(KeyInfo keyInfo, LocalRegion localRegion, boolean allowTom
       PartitionedRegion pr = (PartitionedRegion) localRegion;
       return pr.nonTXGetEntry(keyInfo, false, allowTombstones);
     } finally {
-      localRegion.cache.getTXMgr().resume(tx);
+      localRegion.cache.getTXMgr().internalResume(tx);
     }
   }
 
@@ -72,7 +72,7 @@ public Object findObject(KeyInfo key, LocalRegion r, boolean isCreate, boolean g
       return r.findObjectInSystem(key, isCreate, tx, generateCallbacks, value, disableCopyOnRead,
           preferCD, requestingClient, clientEvent, returnTombstones);
     } finally {
-      r.cache.getTXMgr().resume(tx);
+      r.cache.getTXMgr().internalResume(tx);
     }
   }
 

geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java

@@ -675,26 +675,26 @@ public final TXStateProxy internalSuspend() {
    * @param tx the transaction to activate.
    * @throws IllegalStateException if this thread already has an active transaction
    */
-  public final void resume(TXStateProxy tx) {
+  public void internalResume(TXStateProxy tx) {
     if (tx != null) {
       TransactionId tid = getTransactionId();
       if (tid != null) {
         throw new java.lang.IllegalStateException(
             LocalizedStrings.TXManagerImpl_TRANSACTION_0_ALREADY_IN_PROGRESS
                 .toLocalizedString(tid));
       }
-      if (tx instanceof TXState) {
-        throw new java.lang.IllegalStateException("Found instance of TXState: " + tx);
-      }
       setTXState(tx);
       tx.resume();
-      SystemTimerTask task = this.expiryTasks.remove(tx.getTransactionId());
-      if (task != null) {
-        task.cancel();
-      }
     }
   }
 
+  /**
+   * @deprecated use internalResume instead
+   */
+  public final void resume(TXStateProxy tx) {
+    internalResume(tx);
+  }
+
   public final boolean isClosed() {
     return this.closed;
   }
@@ -1265,7 +1265,7 @@ public void resume(TransactionId transactionId) {
       throw new IllegalStateException(
           LocalizedStrings.TXManagerImpl_UNKNOWN_TRANSACTION_OR_RESUMED.toLocalizedString());
     }
-    resume(txProxy);
+    resumeProxy(txProxy);
   }
 
   public boolean isSuspended(TransactionId transactionId) {
@@ -1278,12 +1278,24 @@ public boolean tryResume(TransactionId transactionId) {
     }
     TXStateProxy txProxy = this.suspendedTXs.remove(transactionId);
     if (txProxy != null) {
-      resume(txProxy);
+      resumeProxy(txProxy);
       return true;
     }
     return false;
   }
 
+  private void resumeProxy(TXStateProxy txProxy) {
+    assert txProxy != null;
+    internalResume(txProxy);
+    SystemTimerTask task = this.expiryTasks.remove(txProxy.getTransactionId());
+    if (task != null) {
+      if (task.cancel()) {
+        GemFireCacheImpl cache = (GemFireCacheImpl) this.cache;
+        cache.purgeCCPTimer();
+      }
+    }
+  }
+
   /**
    * this map keeps track of all the threads that are waiting in
    * {@link #tryResume(TransactionId, long, TimeUnit)} for a particular transactionId

geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java

@@ -1679,11 +1679,11 @@ public Object getEntryForIterator(KeyInfo curr, LocalRegion currRgn, boolean rem
       if (!pr.getBucketPrimary(curr.getBucketId()).equals(pr.cache.getMyId())) {
         // to fix bug 47893 suspend the tx before calling nonTXGetEntry
         final TXManagerImpl txmgr = pr.getGemFireCache().getTXMgr();
-        TransactionId tid = txmgr.suspend();
+        final TXStateProxy tx = txmgr.internalSuspend();
         try {
           return pr.nonTXGetEntry(curr, false, allowTombstones);
         } finally {
-          txmgr.resume(tid);
+          txmgr.internalResume(tx);
         }
       }
     }

geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxy.java

@@ -70,8 +70,8 @@ public interface TXStateProxy extends TXStateInterface {
   public void suspend();
 
   /**
-   * Called by {@link TXManagerImpl#resume(TXStateProxy)} to perform additional tasks required to
-   * resume a transaction
+   * Called by {@link TXManagerImpl#internalResume(TXStateProxy)} to perform additional tasks
+   * required to resume a transaction
    */
   public void resume();
 

geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java

@@ -677,7 +677,7 @@ public int entryCount(LocalRegion localRegion) {
       return getRealDeal(null, localRegion).entryCount(localRegion);
     } finally {
       if (resetTXState) {
-        getTxMgr().resume(txp);
+        getTxMgr().internalResume(txp);
       } else if (txUnlocked) {
         getLock().lock();
       }
@@ -742,7 +742,7 @@ public Object getEntryForIterator(KeyInfo key, LocalRegion currRgn, boolean reme
           allowTombstones);
     } finally {
       if (resetTxState) {
-        getTxMgr().resume(txp);
+        getTxMgr().internalResume(txp);
       }
     }
 
@@ -770,7 +770,7 @@ public Object getKeyForIterator(KeyInfo keyInfo, LocalRegion currRgn, boolean re
           allowTombstones);
     } finally {
       if (resetTxState) {
-        getTxMgr().resume(txp);
+        getTxMgr().internalResume(txp);
       }
     }
   }
@@ -940,7 +940,7 @@ public Set getBucketKeys(LocalRegion localRegion, int bucketId, boolean allowTom
       return getRealDeal(null, localRegion).getBucketKeys(localRegion, bucketId, false);
     } finally {
       if (resetTxState) {
-        getTxMgr().resume(txp);
+        getTxMgr().internalResume(txp);
       }
     }
   }

geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxy.java

@@ -1970,7 +1970,9 @@ public void run2() {
          */
         Object task = _durableExpirationTask.getAndSet(null);
         if (task != null) {
-          ((SystemTimerTask) task).cancel();
+          if (((SystemTimerTask) task).cancel()) {
+            _cache.purgeCCPTimer();
+          }
         }
       }
 
@@ -1988,7 +1990,9 @@ protected void cancelDurableExpirationTask(boolean logMessage) {
             LocalizedStrings.CacheClientProxy_0_CANCELLING_EXPIRATION_TASK_SINCE_THE_CLIENT_HAS_RECONNECTED,
             this));
       }
-      task.cancel();
+      if (task.cancel()) {
+        _cache.purgeCCPTimer();
+      }
     }
   }
 

geode-core/src/main/java/org/apache/geode/pdx/internal/PeerTypeRegistration.java

@@ -642,7 +642,7 @@ private TXStateProxy suspendTX() {
   private void resumeTX(TXStateProxy state) {
     if (state != null) {
       TXManagerImpl txManager = state.getTxMgr();
-      txManager.resume(state);
+      txManager.internalResume(state);
     }
   }
 

geode-core/src/test/java/org/apache/geode/TXExpiryJUnitTest.java

@@ -200,7 +200,7 @@ public void afterRegionDestroy(RegionEvent<String, String> event) {
           final TXManagerImpl txMgrImpl = (TXManagerImpl) this.txMgr;
           TXStateProxy tx = txMgrImpl.internalSuspend();
           exprReg.put("key0", checkVal);
-          txMgrImpl.resume(tx);
+          txMgrImpl.internalResume(tx);
           try {
             this.txMgr.commit();
             fail("Expected CommitConflictException!");