[BACKPORT] EntryProcessor with Predicate should not touch non-matching entries

hazelcast/src/main/java/com/hazelcast/map/impl/operation/EntryOperator.java

@@ -41,6 +41,7 @@
 import com.hazelcast.spi.EventService;
 import com.hazelcast.spi.NodeEngine;
 import com.hazelcast.spi.partition.IPartitionService;
+import com.hazelcast.util.Clock;
 
 import java.util.Map;
 
@@ -88,6 +89,7 @@ public final class EntryOperator {
     private Object newValue;
     private EntryEventType eventType;
     private Data result;
+    private boolean didMatchPredicate;
 
     @SuppressWarnings("checkstyle:executablestatementcount")
     private EntryOperator(MapOperation mapOperation, Object processor, Predicate predicate, boolean collectWanEvents) {
@@ -143,6 +145,7 @@ public EntryOperator init(Data dataKey, Object oldValue, Object newValue, Data r
         this.newValue = newValue;
         this.eventType = eventType;
         this.result = result;
+        this.didMatchPredicate = true;
         return this;
     }
 
@@ -153,7 +156,7 @@ public EntryOperator operateOnKey(Data dataKey) {
             return this;
         }
 
-        oldValue = recordStore.get(dataKey, backup, callerAddress);
+        oldValue = recordStore.get(dataKey, backup, callerAddress, false);
         // predicated entry processors can only be applied to existing entries
         // so if we have a predicate and somehow(due to expiration or split-brain healing)
         // we found value null, we should skip that entry.
@@ -175,6 +178,7 @@ private EntryOperator operateOnKeyValueInternal(Data dataKey, Object oldValue, B
 
         Map.Entry entry = createMapEntry(dataKey, oldValue, locked);
         if (outOfPredicateScope(entry)) {
+            this.didMatchPredicate = false;
             return this;
         }
 
@@ -209,13 +213,20 @@ public Data getResult() {
     }
 
     public EntryOperator doPostOperateOps() {
+        if (!didMatchPredicate) {
+            return this;
+        }
         if (eventType == null) {
             // when event type is null, it means this is a read-only entry processor and not modified entry.
+            onTouched();
             return this;
         }
         switch (eventType) {
-            case ADDED:
             case UPDATED:
+                onTouched();
+                onAddedOrUpdated();
+                break;
+            case ADDED:
                 onAddedOrUpdated();
                 break;
             case REMOVED:
@@ -281,6 +292,14 @@ private void findModificationType(Map.Entry entry) {
         eventType = oldValue == null ? ADDED : UPDATED;
     }
 
+    private void onTouched() {
+        // updates access time if record exists
+        Record record = recordStore.getRecord(dataKey);
+        if (record != null) {
+            recordStore.accessRecord(record, Clock.currentTimeMillis());
+        }
+    }
+
     private void onAddedOrUpdated() {
         if (backup) {
             recordStore.putBackup(dataKey, newValue, NOT_WAN);

hazelcast/src/main/java/com/hazelcast/map/impl/recordstore/AbstractEvictableRecordStore.java

@@ -117,6 +117,11 @@ public boolean isExpirable() {
         return isRecordStoreExpirable();
     }
 
+    @Override
+    public Object get(Data dataKey, boolean backup, Address callerAddress) {
+        return get(dataKey, backup, callerAddress, true);
+    }
+
     /**
      * Intended to put an upper bound to iterations. Used in evictions.
      *
@@ -343,7 +348,8 @@ private void accumulateOrSendExpiredKey(Record record) {
         clearExpiredRecordsTask.tryToSendBackupExpiryOp(this, true);
     }
 
-    protected void accessRecord(Record record, long now) {
+    @Override
+    public void accessRecord(Record record, long now) {
         record.onAccess(now);
         updateStatsOnGet(now);
         setExpirationTime(record);

hazelcast/src/main/java/com/hazelcast/map/impl/recordstore/DefaultRecordStore.java

@@ -508,14 +508,14 @@ public boolean remove(Data key, Object testValue) {
     }
 
     @Override
-    public Object get(Data key, boolean backup, Address callerAddress) {
+    public Object get(Data key, boolean backup, Address callerAddress, boolean touch) {
         checkIfLoaded();
         long now = getNow();
 
         Record record = getRecordOrNull(key, now, backup);
         if (record == null) {
             record = loadRecordOrNull(key, backup, callerAddress);
-        } else {
+        } else if (touch) {
             accessRecord(record, now);
         }
         Object value = record == null ? null : record.getValue();

hazelcast/src/main/java/com/hazelcast/map/impl/recordstore/RecordStore.java

@@ -59,6 +59,17 @@ public interface RecordStore<R extends Record> {
 
     LocalRecordStoreStats getLocalRecordStoreStats();
 
+    /**
+     * Touches the given record, updating its last access time to {@code now} and
+     * maintaining statistics.
+     * <p>
+     * An implementation is not supposed to be thread safe.
+     *
+     * @param record the accessed record
+     * @param now    the current time
+     */
+    void accessRecord(Record record, long now);
+
     String getName();
 
     /**
@@ -129,9 +140,17 @@ public interface RecordStore<R extends Record> {
      * Loads missing keys from map store.
      *
      * @param dataKey key.
-     * @param backup  <code>true</code> if a backup partition, otherwise <code>false</code>.
+     * @param backup  {@code true} if a backup partition, otherwise {@code false}.
+     * @param touch   when {@code true}, if an existing record was found for the given key,
+     *                then its last access time is updated.
      * @return value of an entry in {@link RecordStore}
      */
+    Object get(Data dataKey, boolean backup, Address callerAddress, boolean touch);
+
+    /**
+     * Same as {@link #get(Data, boolean, Address, boolean)} with parameter {@code touch}
+     * set {@code true}.
+     */
     Object get(Data dataKey, boolean backup, Address callerAddress);
 
     /**

hazelcast/src/test/java/com/hazelcast/map/EntryProcessorTest.java

@@ -22,6 +22,7 @@
 import com.hazelcast.config.MapIndexConfig;
 import com.hazelcast.config.MapStoreConfig;
 import com.hazelcast.core.EntryEvent;
+import com.hazelcast.core.EntryView;
 import com.hazelcast.core.ExecutionCallback;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.HazelcastInstanceAware;
@@ -2007,6 +2008,48 @@ public void multiple_entry_with_predicate_operation_returns_empty_response_when_
         }
     }
 
+    // when executing EntryProcessor with predicate via partition scan
+    // entries not matching the predicate should not be touched
+    // see https://github.com/hazelcast/hazelcast/issues/15515
+    @Test
+    public void testEntryProcessorWithPredicate_doesNotTouchNonMatchingEntries() {
+        testEntryProcessorWithPredicate_updatesLastAccessTime(false);
+    }
+
+    @Test
+    public void testEntryProcessorWithPredicate_touchesMatchingEntries() {
+        testEntryProcessorWithPredicate_updatesLastAccessTime(true);
+    }
+
+    private void testEntryProcessorWithPredicate_updatesLastAccessTime(final boolean accessExpected) {
+        Config config = smallInstanceConfig();
+        config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
+        config.getNetworkConfig().getJoin().getTcpIpConfig().setEnabled(false);
+        config.getMapConfig(MAP_NAME)
+              .setTimeToLiveSeconds(60)
+              .setMaxIdleSeconds(30);
+
+        HazelcastInstance member = createHazelcastInstance(config);
+        IMap<String, String> map = member.getMap(MAP_NAME);
+        map.put("testKey", "testValue");
+        EntryView evStart = map.getEntryView("testKey");
+        sleepAtLeastSeconds(2);
+        map.executeOnEntries(new NoOpEntryProcessor(), new Predicate() {
+            @Override
+            public boolean apply(Map.Entry mapEntry) {
+                return accessExpected;
+            }
+        });
+        EntryView evEnd = map.getEntryView("testKey");
+
+        if (accessExpected) {
+            assertTrue("Expiration time should be greater than original one",
+                    evEnd.getExpirationTime() > evStart.getExpirationTime());
+        } else {
+            assertEquals("Expiration time should be the same", evStart.getExpirationTime(), evEnd.getExpirationTime());
+        }
+    }
+
     static class MyData implements Serializable {
         private long lastValue;
 

hazelcast/src/test/java/com/hazelcast/map/MapRemoveAllTest.java

@@ -132,6 +132,21 @@ public void removes_same_number_of_entries_from_owner_and_backup() {
         assertEquals(100, totalBackupEntryCount);
     }
 
+    // see https://github.com/hazelcast/hazelcast/issues/15515
+    @Test
+    public void removeAll_doesNotTouchNonMatchingEntries() {
+        String mapName = "test";
+        IMap<Integer, Integer> map = member.getMap(mapName);
+        for (int i = 0; i < 1000; i++) {
+            map.put(i, i);
+        }
+        long expirationTime = map.getEntryView(1).getExpirationTime();
+
+        map.removeAll(new SqlPredicate("__key >= 100"));
+        assertEquals("Expiration time of non-matching key 1 should be same as original",
+                expirationTime, map.getEntryView(1).getExpirationTime());
+    }
+
     private static final class OddFinderPredicate implements Predicate<Integer, Integer> {
         @Override
         public boolean apply(Map.Entry<Integer, Integer> mapEntry) {