GEODE-6195 putIfAbsent may get a returned value caused by the same operation due to retry

Moving the check for a retried putIfAbsent to be under the RegionEntry
lock.  This allows the operation to propagate throughout the cluster and
allows the client to receive a valid version tag, if concurrency checks
are enabled.

(cherry-picked from 2b2fda6)

Author: bschuchardt

geode-core/src/integrationTest/java/org/apache/geode/cache/RetryPutIfAbsentIntegrationTest.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.cache;
+
+import static org.apache.geode.internal.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import org.apache.geode.cache.query.CacheUtils;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.EventIDHolder;
+import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.internal.cache.VMCachedDeserializable;
+import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
+
+public class RetryPutIfAbsentIntegrationTest {
+
+  Cache cache;
+
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
+  @Test
+  public void duplicatePutIfAbsentIsAccepted() {
+    final String key = "mykey";
+    final String value = "myvalue";
+
+    LocalRegion myregion =
+        (LocalRegion) CacheUtils.getCache().createRegionFactory(RegionShortcut.REPLICATE)
+            .setConcurrencyChecksEnabled(true).create("myregion");
+
+    ClientProxyMembershipID id =
+        new ClientProxyMembershipID(new InternalDistributedMember("localhost", 1));
+    EventIDHolder clientEvent = new EventIDHolder(new EventID(new byte[] {1, 2, 3, 4, 5}, 1, 1));
+    clientEvent.setRegion(myregion);
+    byte[] valueBytes = new VMCachedDeserializable("myvalue", 7).getSerializedValue();
+    System.out.println("first putIfAbsent");
+    Object oldValue =
+        myregion.basicBridgePutIfAbsent("mykey", valueBytes, true, null, id, true, clientEvent);
+    assertEquals(null, oldValue);
+    assertTrue(myregion.containsKey(key));
+
+    myregion.getEventTracker().clear();
+
+    clientEvent = new EventIDHolder(new EventID(new byte[] {1, 2, 3, 4, 5}, 1, 1));
+    clientEvent.setRegion(myregion);
+    clientEvent.setPossibleDuplicate(true);
+    clientEvent.setOperation(Operation.PUT_IF_ABSENT);
+    assertFalse(myregion.getEventTracker().hasSeenEvent(clientEvent));
+
+    System.out.println("second putIfAbsent");
+    oldValue =
+        myregion.basicBridgePutIfAbsent("mykey", valueBytes, true, null, id, true, clientEvent);
+    assertEquals(null, oldValue);
+  }
+
+  @Before
+  public void setUp() throws Exception {
+    CacheUtils.startCache();
+    cache = CacheUtils.getCache();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    CacheUtils.closeCache();
+  }
+
+}

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

@@ -3088,42 +3088,8 @@ void serverPut(EntryEventImpl event, boolean requireOldValue, Object expectedOld
 
   void checkPutIfAbsentResult(EntryEventImpl event, Object value, Object result) {
     if (result != null) {
-      // we may see a non null result possibly due to retry
-      if (event.hasRetried() && putIfAbsentResultHasSameValue(true, value, result)) {
-        if (logger.isDebugEnabled()) {
-          logger.debug("retried putIfAbsent and result is the value to be put,"
-              + " treat as a successful putIfAbsent");
-        }
-      } else {
-        // customers don't see this exception
-        throw new EntryNotFoundException("entry existed for putIfAbsent");
-      }
-    }
-  }
-
-  boolean putIfAbsentResultHasSameValue(boolean isClient, Object valueToBePut, Object result) {
-    if (Token.isInvalid(result) || result == null) {
-      return valueToBePut == null;
-    }
-
-    boolean isCompressedOffHeap =
-        isClient ? false : getAttributes().getOffHeap() && getAttributes().getCompressor() != null;
-    return ValueComparisonHelper.checkEquals(valueToBePut, result, isCompressedOffHeap, getCache());
-  }
-
-  boolean bridgePutIfAbsentResultHasSameValue(byte[] valueToBePut, boolean isValueToBePutObject,
-      Object result) {
-    if (Token.isInvalid(result) || result == null) {
-      return valueToBePut == null;
-    }
-
-    boolean isCompressedOffHeap =
-        getAttributes().getOffHeap() && getAttributes().getCompressor() != null;
-    if (isValueToBePutObject) {
-      return ValueComparisonHelper.checkEquals(EntryEventImpl.deserialize(valueToBePut), result,
-          isCompressedOffHeap, getCache());
+      throw new EntryNotFoundException("entry existed for putIfAbsent");
     }
-    return ValueComparisonHelper.checkEquals(valueToBePut, result, isCompressedOffHeap, getCache());
   }
 
   /**
@@ -10923,22 +10889,14 @@ public Object putIfAbsent(Object key, Object value, Object callbackArgument) {
 
     try {
       if (generateEventID()) {
-        event.setNewEventId(this.cache.getDistributedSystem());
+        event.setNewEventId(cache.getDistributedSystem());
       }
       final Object oldValue = null;
       final boolean ifNew = true;
       final boolean ifOld = false;
       final boolean requireOldValue = true;
       if (!basicPut(event, ifNew, ifOld, oldValue, requireOldValue)) {
-        Object result = event.getOldValue();
-        if (event.isPossibleDuplicate() && putIfAbsentResultHasSameValue(false, value, result)) {
-          if (logger.isDebugEnabled()) {
-            logger.debug("possible duplicate putIfAbsent event and result is the value to be put,"
-                + " treat this as a successful putIfAbsent");
-          }
-          return null;
-        }
-        return result;
+        return event.getOldValue();
       } else {
         if (!getDataView().isDeferredStats()) {
           getCachePerfStats().endPut(startPut, false);
@@ -11138,6 +11096,7 @@ public Object basicBridgePutIfAbsent(final Object key, Object value, boolean isO
 
       // if this is a replayed operation we may already have a version tag
       event.setVersionTag(clientEvent.getVersionTag());
+      event.setPossibleDuplicate(clientEvent.isPossibleDuplicate());
 
       // Set the new value to the input byte[] if it isn't null
       if (value != null) {
@@ -11176,19 +11135,6 @@ public Object basicBridgePutIfAbsent(final Object key, Object value, boolean isO
         clientEvent.setVersionTag(event.getVersionTag());
         clientEvent.isConcurrencyConflict(event.isConcurrencyConflict());
       } else {
-        if (value != null) {
-          assert (value instanceof byte[]);
-        }
-        if (event.isPossibleDuplicate()
-            && bridgePutIfAbsentResultHasSameValue((byte[]) value, isObject, oldValue)) {
-          // result is possibly due to the retry
-          if (logger.isDebugEnabled()) {
-            logger.debug("retried putIfAbsent and got oldValue as the value to be put,"
-                + " treat this as a successful putIfAbsent");
-          }
-          return null;
-        }
-
         if (oldValue == null) {
           // fix for 42189, putIfAbsent on server can return null if the
           // operation was not performed (oldValue in cache was null).

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

@@ -127,6 +127,13 @@ public void stop() {
     }
   }
 
+  @Override
+  public void clear() {
+    recordedEvents.clear();
+    recordedBulkOps.clear();
+    recordedBulkOpVersionTags.clear();
+  }
+
   @Override
   public Map<ThreadIdentifier, EventSequenceNumberHolder> getState() {
     Map<ThreadIdentifier, EventSequenceNumberHolder> result = new HashMap<>(recordedEvents.size());

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

@@ -133,4 +133,9 @@ void recordState(InternalDistributedMember provider,
    */
   boolean isInitialImageProvider(DistributedMember mbr);
 
+  /**
+   * clear the tracker
+   */
+  void clear();
+
 }

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

@@ -49,6 +49,11 @@ public void stop() {
 
   }
 
+  @Override
+  public void clear() {
+    // nothing to clear
+  }
+
   @Override
   public Map<ThreadIdentifier, EventSequenceNumberHolder> getState() {
     return null;

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

@@ -18,6 +18,8 @@
 
 import java.util.Set;
 
+import org.apache.logging.log4j.Logger;
+
 import org.apache.geode.cache.CacheWriter;
 import org.apache.geode.cache.DiskAccessException;
 import org.apache.geode.cache.Operation;
@@ -27,22 +29,27 @@
 import org.apache.geode.internal.cache.RegionClearedException;
 import org.apache.geode.internal.cache.RegionEntry;
 import org.apache.geode.internal.cache.Token;
+import org.apache.geode.internal.cache.ValueComparisonHelper;
 import org.apache.geode.internal.cache.entries.AbstractRegionEntry;
 import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
 import org.apache.geode.internal.cache.versions.VersionTag;
 import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
+import org.apache.geode.internal.logging.LogService;
 import org.apache.geode.internal.offheap.OffHeapHelper;
 import org.apache.geode.internal.offheap.ReferenceCountHelper;
 import org.apache.geode.internal.offheap.annotations.Released;
 import org.apache.geode.internal.offheap.annotations.Unretained;
 import org.apache.geode.internal.sequencelog.EntryLogger;
 
 public class RegionMapPut extends AbstractRegionMapPut {
+  protected static final Logger logger = LogService.getLogger();
+
   private final CacheModificationLock cacheModificationLock;
   private final EntryEventSerialization entryEventSerialization;
   private final boolean ifNew;
   private final boolean ifOld;
   private final boolean overwriteDestroyed;
+  private boolean overwritePutIfAbsent;
   private final boolean requireOldValue;
   private final boolean retrieveOldValueForDelta;
   private final boolean replaceOnClient;
@@ -104,6 +111,10 @@ boolean isReplaceOnClient() {
     return replaceOnClient;
   }
 
+  boolean isOverwritePutIfAbsent() {
+    return overwritePutIfAbsent;
+  }
+
   boolean isCacheWrite() {
     return cacheWrite;
   }
@@ -214,6 +225,9 @@ private void setOldValueEvenIfFaultedOut() {
   protected void unsetOldValueForDelta() {
     OffHeapHelper.release(getOldValueForDelta());
     setOldValueForDelta(null);
+    if (isOverwritePutIfAbsent()) {
+      getEvent().setOldValue(null);
+    }
   }
 
   @Override
@@ -396,12 +410,41 @@ private boolean checkUninitializedRegionPreconditions() {
   private boolean checkCreatePreconditions() {
     if (isIfNew()) {
       if (!getRegionEntry().isDestroyedOrRemoved()) {
+        // retain the version stamp of the existing entry for use in processing failures
+        EntryEventImpl event = getEvent();
+        if (getOwner().getConcurrencyChecksEnabled() &&
+            event.getOperation() == Operation.PUT_IF_ABSENT &&
+            !event.hasValidVersionTag() &&
+            event.isPossibleDuplicate()) {
+          Object retainedValue = getRegionEntry().getValueRetain(getOwner());
+          try {
+            if (ValueComparisonHelper.checkEquals(retainedValue,
+                getEvent().getRawNewValue(),
+                isCompressedOffHeap(event), getOwner().getCache())) {
+              if (logger.isDebugEnabled()) {
+                logger.debug("retried putIfAbsent found same value already in cache "
+                    + "- allowing the operation.  entry={}; event={}", getRegionEntry(),
+                    getEvent());
+              }
+              this.overwritePutIfAbsent = true;
+              return true;
+            }
+          } finally {
+            OffHeapHelper.release(retainedValue);
+          }
+        }
         return false;
       }
     }
     return true;
   }
 
+
+  private boolean isCompressedOffHeap(EntryEventImpl event) {
+    return event.getRegion().getAttributes().getOffHeap()
+        && event.getRegion().getAttributes().getCompressor() != null;
+  }
+
   private boolean checkExpectedOldValuePrecondition() {
     // replace is propagated to server, so no need to check
     // satisfiesOldValue on client
@@ -411,7 +454,7 @@ private boolean checkExpectedOldValuePrecondition() {
       // We already called setOldValueInEvent so the event will have the old value.
       @Unretained
       Object v = event.getRawOldValue();
-      // Note that v will be null instead of INVALID because setOldValue
+      // Note that v will be null instead of INVALID because setOldValue`
       // converts INVALID to null.
       // But checkExpectedOldValue handle this and says INVALID equals null.
       if (!AbstractRegionEntry.checkExpectedOldValue(getExpectedOldValue(), v, event.getRegion())) {