Fix possible NPE during remove-if-same operation

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

@@ -508,21 +508,24 @@ public boolean remove(Data key, Object testValue) {
         } else {
             oldValue = record.getValue();
         }
+
         if (valueComparator.isEqual(testValue, oldValue, serializationService)) {
             mapServiceContext.interceptRemove(name, oldValue);
-            removeIndex(record);
             mapDataStore.remove(key, now);
-            onStore(record);
-            mutationObserver.onRemoveRecord(record.getKey(), record);
-            storage.removeRecord(record);
+
+            if (record != null) {
+                removeIndex(record);
+                onStore(record);
+                mutationObserver.onRemoveRecord(key, record);
+                storage.removeRecord(record);
+            }
             removed = true;
         }
         return removed;
     }
 
     @Override
-    public Object get(Data key, boolean backup, Address
-            callerAddress) {
+    public Object get(Data key, boolean backup, Address callerAddress) {
         checkIfLoaded();
         long now = getNow();
 

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

@@ -20,6 +20,8 @@
 import com.hazelcast.map.impl.record.Record;
 import com.hazelcast.nio.serialization.Data;
 
+import javax.annotation.Nonnull;
+
 /**
  * Interface for observing {@link RecordStore} mutations
  *
@@ -37,7 +39,7 @@ public interface RecordStoreMutationObserver<R extends Record> {
      * @param key    The key of the record
      * @param record The record
      */
-    void onPutRecord(Data key, R record);
+    void onPutRecord(@Nonnull Data key, @Nonnull R record);
 
     /**
      * Called when a new record is added to the {@link RecordStore} due
@@ -46,7 +48,7 @@ public interface RecordStoreMutationObserver<R extends Record> {
      * @param key    The key of the record
      * @param record The record
      */
-    void onReplicationPutRecord(Data key, R record);
+    void onReplicationPutRecord(@Nonnull Data key, @Nonnull R record);
 
     /**
      * Called when a new record is updated in the observed {@link RecordStore}
@@ -55,31 +57,31 @@ public interface RecordStoreMutationObserver<R extends Record> {
      * @param record   The record
      * @param newValue The new value of the record
      */
-    void onUpdateRecord(Data key, R record, Object newValue);
+    void onUpdateRecord(@Nonnull Data key, @Nonnull R record, Object newValue);
 
     /**
      * Called when a record is removed from the observed {@link RecordStore}
      *
      * @param key    The key of the record
      * @param record The record
      */
-    void onRemoveRecord(Data key, R record);
+    void onRemoveRecord(@Nonnull Data key, R record);
 
     /**
      * Called when a record is evicted from the observed {@link RecordStore}
      *
      * @param key    The key of the record
      * @param record The record
      */
-    void onEvictRecord(Data key, R record);
+    void onEvictRecord(@Nonnull Data key, @Nonnull R record);
 
     /**
      * Called when a record is loaded into the observed {@link RecordStore}
      *
-     * @param key      The key of the record
-     * @param record   The record
+     * @param key    The key of the record
+     * @param record The record
      */
-    void onLoadRecord(Data key, R record);
+    void onLoadRecord(@Nonnull Data key, @Nonnull R record);
 
     /**
      * Called when the observed {@link RecordStore} is being destroyed.

hazelcast/src/test/java/com/hazelcast/map/impl/operation/RemoveIfSameOperationTest.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright (c) 2008-2019, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.map.impl.operation;
+
+import com.hazelcast.config.Config;
+import com.hazelcast.config.MapStoreConfig;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.map.MapStoreAdapter;
+import com.hazelcast.test.HazelcastSerialClassRunner;
+import com.hazelcast.test.HazelcastTestSupport;
+import com.hazelcast.test.annotation.ParallelJVMTest;
+import com.hazelcast.test.annotation.QuickTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import static junit.framework.TestCase.assertEquals;
+
+@RunWith(HazelcastSerialClassRunner.class)
+@Category({QuickTest.class, ParallelJVMTest.class})
+public class RemoveIfSameOperationTest extends HazelcastTestSupport {
+
+    @Test
+    public void removes_key_if_it_is_mapped_to_given_value() {
+        int key = 1;
+        String value = "value";
+
+        ConcurrentMap<Integer, String> store = new ConcurrentHashMap<>();
+        store.put(key, value);
+
+        String mapName = "default";
+        MapStoreConfig mapStoreConfig = new MapStoreConfig();
+        mapStoreConfig.setEnabled(true);
+        mapStoreConfig.setImplementation(new MapStoreAdapter<Integer, String>() {
+
+            @Override
+            public String load(Integer key) {
+                return store.get(key);
+            }
+
+            @Override
+            public void delete(Integer key) {
+                store.remove(key);
+            }
+        });
+
+        Config config = getConfig();
+        config.getMapConfig(mapName).setMapStoreConfig(mapStoreConfig);
+
+        HazelcastInstance node = createHazelcastInstance(config);
+
+        node.getMap(mapName).remove(key, value);
+
+        assertEquals(0, store.size());
+    }
+}