fixed race condition in old entries cleaner thread deleting in-use sn…

…apshot entries

- use a lock for entire duration of getting the active transactions as well as checking if
  an entry is in-use by a transaction instead of just taking a snapshot of the active
  transactions; this fixes a race condition where the old entries cleaner thread deleting
  in-use entries by an active transaction started just after the active transaction list
  was obtained
- fixed assertion error in changePassword when the given password or old password is null
- fixed occasional failures in BugsDUnit, ClientServer2DUnit, MVCCDUnit and HeapThresholdDUnit
- fixed occasional failures in Bugs3Test when run in the full parallel suite
- workaround occasional failure in StatementStatsDUnit since those stats are never used
  by SnappyData

gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/GemFireCacheImpl.java

@@ -888,18 +888,20 @@ class OldEntriesCleanerThread implements Runnable {
      */
     private long refreshRunningTXs(ArrayList<TXState> runningTXs, long txListVersion) {
       final TXManagerImpl txMgr = getTxManager();
-      final long newTxListVersion = txMgr.hostTXStatesVersion.get();
-      if (txListVersion != newTxListVersion) {
-        if (txListVersion >= 0) runningTXs.clear();
-        Collection<TXStateProxy> txProxies = getTxManager().getHostedTransactionsInProgress();
-        for (TXStateProxy txProxy : txProxies) {
-          TXState txState = txProxy.getLocalTXState();
-          if (txState != null && txState.isSnapshot() && !txState.isClosed()) {
-            runningTXs.add(txState);
+      synchronized (txMgr.hostedTXStatesLock) {
+        final long newTxListVersion = txMgr.getHostedTXStateVersion();
+        if (txListVersion != newTxListVersion) {
+          if (txListVersion >= 0) runningTXs.clear();
+          Collection<TXStateProxy> txProxies = txMgr.getHostedTransactionsInProgress();
+          for (TXStateProxy txProxy : txProxies) {
+            TXState txState = txProxy.getLocalTXState();
+            if (txState != null && txState.isSnapshot() && !txState.isClosed()) {
+              runningTXs.add(txState);
+            }
           }
         }
+        return newTxListVersion;
       }
-      return newTxListVersion;
     }
 
     public void run() {
@@ -931,11 +933,16 @@ public void run() {
                   } else {
                     txListVersion = refreshRunningTXs(runningTXs, txListVersion);
                     if (notRequired(region, re, null, runningTXs)) {
+                      if (SNAPSHOT_DEBUG || getLoggerI18n().fineEnabled()) {
+                        getLoggerI18n().info(LocalizedStrings.DEBUG,
+                            "OldEntriesCleanerThread: Removing the entry " + re);
+                      }
                       removeEntry(regionEntryMap, re, region);
                     }
                   }
                 } else {
-                  BlockingQueue<RegionEntry> oldEntriesQueue = (BlockingQueue<RegionEntry>) oldEntries;
+                  @SuppressWarnings("unchecked")
+                  BlockingQueue<RegionEntry> oldEntriesQueue = (BlockingQueue<RegionEntry>)oldEntries;
                   for (RegionEntry re : oldEntriesQueue) {
                     // update in progress guards against the race where oldEntry and
                     // entry in region have same version for brief period
@@ -946,7 +953,7 @@ public void run() {
                       if (notRequired(region, re, oldEntriesQueue, runningTXs)) {
                         if (SNAPSHOT_DEBUG || getLoggerI18n().fineEnabled()) {
                           getLoggerI18n().info(LocalizedStrings.DEBUG,
-                                  "OldEntriesCleanerThread : Removing the entry " + re);
+                              "OldEntriesCleanerThread: Removing the entry from queue " + re);
                         }
                         // continue if some explicit call removed the entry
                         if (!oldEntriesQueue.remove(re)) continue;
@@ -1160,8 +1167,9 @@ public Object removeValue(Object key, Object value, Object existingValue,
       if (value != null
           && (value == NO_OBJECT_TOKEN || ((existingValue == value)
           && ((BlockingQueue)existingValue).isEmpty()))) {
-        if (getInstance().getLoggerI18n().fineEnabled()) {
-          getInstance().getLoggerI18n().info(LocalizedStrings.DEBUG, "Removing queue for key " + key);
+        final GemFireCacheImpl cache = getInstance();
+        if (cache != null && cache.getLoggerI18n().fineEnabled()) {
+          cache.getLoggerI18n().info(LocalizedStrings.DEBUG, "Removing queue for key " + key);
         }
         return null;
       }

gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/TXManagerImpl.java

@@ -22,7 +22,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.LockSupport;
 import javax.transaction.Transaction;
@@ -116,9 +115,10 @@ public final class TXManagerImpl implements CacheTransactionManager,
 
   private boolean closed = false;
 
-  // whenever the hostedTXStates is changed, the hostTXStatesVersion must be incremented
+  // whenever the hostedTXStates is changed, the hostedTXStatesVersion must be incremented
   private final CustomEntryConcurrentHashMap<TXId, TXStateProxy> hostedTXStates;
-  final AtomicLong hostTXStatesVersion;
+  final Object hostedTXStatesLock = new Object();
+  private long hostedTXStatesVersion;
 
   private static final ConcurrentTHashSet<TXContext> hostedTXContexts =
       new ConcurrentTHashSet<>(16, 128);
@@ -465,10 +465,9 @@ private final TXStateProxy newTXStateProxy(final TXId txId,
       final IsolationLevel isolationLevel, final boolean isJTA,
       final EnumSet<TransactionFlag> flags, final boolean initLocalTXState,
       final MapResult result) {
-    TXStateProxy proxy = GemFireCacheImpl.FactoryStatics.txStateProxyFactory.newTXStateProxy(
+    incrementHostedTXStateVersion();
+    return GemFireCacheImpl.FactoryStatics.txStateProxyFactory.newTXStateProxy(
         this, txId, isolationLevel, isJTA, flags, initLocalTXState);
-    this.hostTXStatesVersion.getAndIncrement();
-    return proxy;
   }
 
   /**
@@ -553,7 +552,7 @@ public final Object removeValue(final Object key, Object value,
             txId.shortToString() + ": removing from hosted list with commit="
                 + commit);
       }
-      hostTXStatesVersion.getAndIncrement();
+      incrementHostedTXStateVersion();
       // always remove from map at this point; callback is only to atomically
       // add to finishedTXStates in postRemove
       return null;
@@ -830,7 +829,7 @@ public TXManagerImpl(CachePerfStats cachePerfStats, LogWriterI18n logWriter,
     this.hostedTXStates = new CustomEntryConcurrentHashMap<>(
         128, CustomEntryConcurrentHashMap.DEFAULT_LOAD_FACTOR,
         TXMAP_CONCURRENCY);
-    this.hostTXStatesVersion = new AtomicLong();
+    this.hostedTXStatesVersion = 0;
     this.suspendedTXs = new ConcurrentHashMap<TXId, TXStateInterface>();
     this.finishedTXStates = new TXFinishedMap(cache.getDistributedSystem(),
         cache.getCancelCriterion());
@@ -993,12 +992,15 @@ public final TXStateProxy beginTX(final TXManagerImpl.TXContext context,
       txId = TXId.newTXId(this.cache);
     }
 
-    final TXStateProxy txState = this.hostedTXStates.create(txId,
-        txStateProxyCreator, isolationLevel, txFlags, false);
-    context.setTXState(txState);
-    // For snapshot isolation, create tx state at the beginning
-    if (txState.isSnapshot()) {
-      txState.getTXStateForRead();
+    final TXStateProxy txState;
+    synchronized (this.hostedTXStatesLock) {
+      txState = this.hostedTXStates.create(txId,
+          txStateProxyCreator, isolationLevel, txFlags, false);
+      context.setTXState(txState);
+      // For snapshot isolation, create tx state at the beginning
+      if (txState.isSnapshot()) {
+        txState.getTXStateForRead();
+      }
     }
 
     return txState;
@@ -1025,8 +1027,10 @@ public final TXStateProxy resumeTX(final TXManagerImpl.TXContext context,
     if (txState != null) {
       return txState;
     }
-    txState = this.hostedTXStates.create(txId,
-        txStateProxyCreator, isolationLevel, txFlags, false);
+    synchronized (this.hostedTXStatesLock) {
+      txState = this.hostedTXStates.create(txId,
+          txStateProxyCreator, isolationLevel, txFlags, false);
+    }
     // context.setTXState(txState);
     return txState;
   }
@@ -1041,8 +1045,11 @@ public TXStateProxy beginJTA() {
     checkClosed();
     final TXContext context = getOrCreateTXContext();
     final TXId txId = TXId.newTXId(this.cache);
-    final TXStateProxy txState = this.hostedTXStates.create(txId,
-        txStateJTACreator, IsolationLevel.DEFAULT, null, false);
+    final TXStateProxy txState;
+    synchronized (this.hostedTXStatesLock) {
+      txState = this.hostedTXStates.create(txId,
+          txStateJTACreator, IsolationLevel.DEFAULT, null, false);
+    }
 
     context.setTXState(txState);
     return txState;
@@ -1690,6 +1697,24 @@ public final void unmasquerade(final TXContext context,
     }
   }
 
+  /**
+   * For package-internal use only.
+   * Caller should hold the lock on {@link #hostedTXStatesLock}.
+   */
+  final void incrementHostedTXStateVersion() {
+    assert Thread.holdsLock(this.hostedTXStatesLock);
+    this.hostedTXStatesVersion++;
+  }
+
+  /**
+   * For package-internal use only.
+   * Caller should hold the lock on {@link #hostedTXStatesLock}.
+   */
+  final long getHostedTXStateVersion() {
+    assert Thread.holdsLock(this.hostedTXStatesLock);
+    return this.hostedTXStatesVersion;
+  }
+
   /**
    * Cleanup the remote txState after commit and rollback
    * @param txId
@@ -1703,15 +1728,18 @@ public final TXStateProxy removeHostedTXState(TXId txId, Boolean commit) {
    * Cleanup the remote txState subject to given condition atomically during
    * remove.
    *
-   * NOTE: the provided MapCallback should increment the {@link TXManagerImpl#hostTXStatesVersion}
+   * NOTE: the provided MapCallback should increment the {@link TXManagerImpl#hostedTXStatesVersion}
    * when removal is given a go ahead (i.e. just before removeValue returns null).
    */
   public final <C, P> TXStateProxy removeHostedTXState(TXId txId,
       MapCallback<TXId, TXStateProxy, C, P> condition, C context,
       P removeParams) {
-    final TXStateProxy tx = this.hostedTXStates.remove(txId, condition,
-        context, removeParams);
-    getCache().removeTXId(txId);
+    final TXStateProxy tx;
+    synchronized (this.hostedTXStatesLock) {
+      tx = this.hostedTXStates.remove(txId, condition,
+          context, removeParams);
+      getCache().removeTXId(txId);
+    }
     if (tx != null) {
       if (TXStateProxy.LOG_FINE) {
         getLogger().info(LocalizedStrings.DEBUG, "TX removed: " + tx);
@@ -2013,8 +2041,12 @@ public TXStateProxy getHostedTXState(TXId txId) {
 
   public TXStateProxy getOrCreateHostedTXState(final TXId txId,
       final LockingPolicy lockingPolicy, final boolean checkFinishTX) {
-    return this.hostedTXStates.create(txId,
-        txStateCreator, lockingPolicy, checkFinishTX, true);
+    final TXStateProxy proxy;
+    synchronized (this.hostedTXStatesLock) {
+      proxy = this.hostedTXStates.create(txId,
+          txStateCreator, lockingPolicy, checkFinishTX, true);
+    }
+    return proxy;
   }
 
   /**

gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/TXStateProxy.java

@@ -562,7 +562,7 @@ public final Object removeValue(final Object key, Object value,
               txProxy.txId.shortToString()
               + ": removing from hosted list since it is empty");
         }
-        txProxy.getTxMgr().hostTXStatesVersion.getAndIncrement();
+        txProxy.getTxMgr().incrementHostedTXStateVersion();
         return null;
       }
       else {
@@ -3478,7 +3478,6 @@ public final TXState getTXStateForRead() {
     }
   }
 
-
   private final TXState createTXState(boolean checkTX) {
     final TXState localState;
     this.lock.lock();

gemfirexd/core/src/main/java/com/pivotal/gemfirexd/internal/engine/db/FabricDatabase.java

@@ -543,9 +543,9 @@ synchronized public void postCreate(
       }
     } catch (Throwable t) {
       try {
-
         if (logger != null) {
-          logger.warning("got throwable: " + t.getMessage() + " calling shut down", t);
+          logger.warning("FabricDatabase.postCreate got throwable: " + t.getMessage() +
+              " calling shut down", t);
         }
         Monitor.getMonitor().shutdown();
       } catch (CancelException ce) {
@@ -575,7 +575,6 @@ synchronized public void postCreate(
           Attribute.GFXD_DBNAME);
     }
 
-
     for (GfxdSystemProcedureMessage msg : postMsgs) {
       if (msg.getSysProcMethod().isOffHeapMethod()
         && this.memStore.getGemFireCache().getOffHeapStore() == null) {

gemfirexd/core/src/main/java/com/pivotal/gemfirexd/internal/engine/ddl/catalog/messages/GfxdSystemProcedureMessage.java

@@ -315,8 +315,8 @@ public void processMessage(Object[] params, DistributedMember sender)
           throws StandardException {
 
         assert String.class.isInstance(params[0])
-            && String.class.isInstance(params[1])
-            && String.class.isInstance(params[2]);
+            && (String.class.isInstance(params[1]) || params[1] == null)
+            && (String.class.isInstance(params[2]) || params[2] == null);
 
         String key = (String)params[0];
         String value = (String)params[2];
@@ -1929,6 +1929,8 @@ String getSQLStatement(Object[] params) throws StandardException {
     },
     ;
 
+    static final SysProcMethod[] values = values();
+
     static final int[][] tableActions = new int[][] {
         { TablePrivilegeInfo.SELECT_ACTION, Authorizer.SELECT_PRIV },
         { TablePrivilegeInfo.INSERT_ACTION, Authorizer.INSERT_PRIV },
@@ -2195,7 +2197,7 @@ public void fromData(DataInput in)
     this.connId = in.readLong();
     this.ddlId = in.readLong();
     int ordinal = in.readByte();
-    this.procMethod = SysProcMethod.values()[ordinal];
+    this.procMethod = SysProcMethod.values[ordinal];
     this.params = this.procMethod.readParams(in, flags);
     this.initialDDLReplayInProgress =
         (flags & INITIAL_DDL_REPLAY_IN_PROGRESS) != 0;

gemfirexd/tools/src/dunit/java/com/pivotal/gemfirexd/BackwardCompatabilityTestBase.java

@@ -34,6 +34,7 @@
 import com.gemstone.gemfire.internal.shared.NativeCalls;
 import com.pivotal.gemfirexd.BackwardCompatabilityDUnit.ClientRun;
 import com.pivotal.gemfirexd.BackwardCompatabilityDUnit.ProductClient;
+import com.pivotal.gemfirexd.internal.engine.Misc;
 import com.pivotal.gemfirexd.tools.GfxdUtilLauncher;
 import io.snappydata.test.dunit.Host;
 import io.snappydata.test.dunit.SerializableCallable;
@@ -122,7 +123,7 @@ public void setUp() throws Exception {
   @Override
   public void tearDown2() throws Exception {
     System.clearProperty("gemfirexd.thrift-default");
-    ClientSharedUtils.setThriftDefault(true);
+    ClientSharedUtils.setThriftDefault(false);
     super.tearDown2();
     final String workingDir = getSysDirName();
     if (currentListIdx >= 0 && currentVersIdx >= 0) {

gemfirexd/tools/src/dunit/java/com/pivotal/gemfirexd/ClientServer2DUnit.java

@@ -266,7 +266,7 @@ public void testInitialScripts() throws Exception {
     } catch (RMIException ex) {
       if (ex.getCause() instanceof SQLException) {
         SQLException sqlEx = (SQLException)ex.getCause();
-        if (!"XJ040".equals(sqlEx.getSQLState())) {
+        if (!"XJ040".equals(sqlEx.getSQLState()) && !"42X05".equals(sqlEx.getSQLState())) {
           throw ex;
         } else {
           // Explicitly delete the newly timestamped persistent file.
@@ -303,7 +303,7 @@ public void testInitialScripts() throws Exception {
     } catch (RMIException ex) {
       if (ex.getCause() instanceof SQLException) {
         SQLException sqlEx = (SQLException)ex.getCause();
-        if (!"XJ040".equals(sqlEx.getSQLState())) {
+        if (!"XJ040".equals(sqlEx.getSQLState()) && !"42X05".equals(sqlEx.getSQLState())) {
           throw ex;
         } else {
           // Explicitly delete the newly timestamped persistent file.

gemfirexd/tools/src/dunit/java/com/pivotal/gemfirexd/ClientServerDUnit.java

@@ -48,6 +48,7 @@
 import com.pivotal.gemfirexd.internal.engine.GemFireXDQueryObserverAdapter;
 import com.pivotal.gemfirexd.internal.engine.GemFireXDQueryObserverHolder;
 import com.pivotal.gemfirexd.internal.engine.GfxdConstants;
+import com.pivotal.gemfirexd.internal.engine.Misc;
 import com.pivotal.gemfirexd.internal.engine.ddl.resolver.GfxdPartitionByExpressionResolver;
 import com.pivotal.gemfirexd.internal.engine.jdbc.GemFireXDRuntimeException;
 import com.pivotal.gemfirexd.internal.engine.store.GemFireStore;
@@ -1834,6 +1835,7 @@ public void testNetworkClientFailoverWithCurrentSchemaSetting() throws Exception
    */
   public void testNetworkClientLoadBalancing() throws Exception {
     // always use thrift for this test
+    final int numVMs = Host.getHost(0).getVMCount();
     SerializableRunnable setThrift = new SerializableRunnable() {
       @Override
       public void run() {
@@ -1843,7 +1845,7 @@ public void run() {
       }
     };
     setThrift.run();
-    for (int i = 0; i <= 3; i++) {
+    for (int i = 0; i < numVMs; i++) {
       Host.getHost(0).getVM(i).invoke(setThrift);
     }
     try {
@@ -1858,7 +1860,7 @@ public void run() {
         }
       };
       resetThrift.run();
-      for (int i = 0; i <= 3; i++) {
+      for (int i = 0; i < numVMs; i++) {
         Host.getHost(0).getVM(i).invoke(resetThrift);
       }
     }

gemfirexd/tools/src/dunit/java/com/pivotal/gemfirexd/ddl/BugsDUnit.java

@@ -2907,7 +2907,6 @@ public void testBug47289() throws Exception {
       for (int i = 1; i < 20; ++i) {
         psSec.setInt(1, i);
         psSec.setString(2, getSymbol(1, 6) + "_" + i);
-        psSec.setString(2, getSymbol(1, 8));
         psSec.setString(3, exchanges[i % 7]);
         psSec.setInt(4, 50);
         psSec.executeUpdate();
@@ -2935,20 +2934,18 @@ public void testBug47289() throws Exception {
         st.execute("alter table trade.companies add constraint comp_fk foreign key (symbol, exchange) "
           + "references trade.securities (symbol, exchange) on delete restrict");
         fail("FK constraint addition should fail");
-      }catch(SQLException sqle) {
-        if(sqle.getSQLState().equals("0A000")) {
+      } catch (SQLException sqle) {
+        if (sqle.getSQLState().equals("0A000") || sqle.getSQLState().equals("23505")) {
           //Ok
-        }else {
+        } else {
           throw sqle;
         }
       }
     } finally {
       TestUtil.shutDown();
     }
-
   }
 
-
   public void testBug47289_1() throws Exception {
 
     // Start two server VMs