Fix txn near cache

hazelcast/src/main/java/com/hazelcast/collection/impl/txncollection/CollectionTransactionLogRecord.java

@@ -64,6 +64,16 @@ public Operation newCommitOperation() {
         return new CollectionCommitOperation(partitionId, name, serviceName, operationList);
     }
 
+    @Override
+    public void onCommitSuccess() {
+        // NOP
+    }
+
+    @Override
+    public void onCommitFailure() {
+        // NOP
+    }
+
     @Override
     public Operation newRollbackOperation() {
         long[] itemIds = createItemIdArray();

hazelcast/src/main/java/com/hazelcast/internal/cluster/impl/ClusterStateTransactionLogRecord.java

@@ -49,7 +49,7 @@ public ClusterStateTransactionLogRecord() {
     }
 
     public ClusterStateTransactionLogRecord(ClusterStateChange stateChange, Address initiator, Address target,
-            String txnId, long leaseTime, int partitionStateVersion, boolean isTransient) {
+                                            String txnId, long leaseTime, int partitionStateVersion, boolean isTransient) {
         Preconditions.checkNotNull(stateChange);
         Preconditions.checkNotNull(initiator);
         Preconditions.checkNotNull(target);
@@ -85,6 +85,16 @@ public Operation newRollbackOperation() {
         return new RollbackClusterStateOp(initiator, txnId);
     }
 
+    @Override
+    public void onCommitSuccess() {
+        // NOP
+    }
+
+    @Override
+    public void onCommitFailure() {
+        // NOP
+    }
+
     @Override
     public Address getTarget() {
         return target;

hazelcast/src/main/java/com/hazelcast/internal/nearcache/NearCachingHook.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c) 2008-2020, 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.internal.nearcache;
+
+
+import com.hazelcast.nio.serialization.Data;
+
+/**
+ * Hook to be used by near cache enabled proxy objects.
+ *
+ * With this hook, you can implement needed logic
+ * for truly invalidate/populate local near cache.
+ */
+public interface NearCachingHook<K, V> {
+
+    NearCachingHook EMPTY_HOOK = new NearCachingHook() {
+
+        @Override
+        public void beforeRemoteCall(Object key, Data keyData,
+                                     Object value, Data valueData) {
+        }
+
+        @Override
+        public void onRemoteCallSuccess() {
+        }
+
+        @Override
+        public void onRemoteCallFailure() {
+
+        }
+    };
+
+    void beforeRemoteCall(K key, Data keyData, V value, Data valueData);
+
+    void onRemoteCallSuccess();
+
+    void onRemoteCallFailure();
+}

hazelcast/src/main/java/com/hazelcast/map/impl/tx/MapTransactionLogRecord.java

@@ -16,6 +16,7 @@
 
 package com.hazelcast.map.impl.tx;
 
+import com.hazelcast.internal.nearcache.NearCachingHook;
 import com.hazelcast.map.impl.MapDataSerializerHook;
 import com.hazelcast.map.impl.MapRecordKey;
 import com.hazelcast.nio.ObjectDataInput;
@@ -39,10 +40,15 @@ public class MapTransactionLogRecord implements TransactionLogRecord {
     private String ownerUuid;
     private Operation op;
 
+    private transient NearCachingHook nearCachingHook = NearCachingHook.EMPTY_HOOK;
+
     public MapTransactionLogRecord() {
     }
 
-    public MapTransactionLogRecord(String name, Data key, int partitionId, Operation op, long version, String ownerUuid) {
+    public MapTransactionLogRecord(String name, Data key, int partitionId,
+                                   Operation op, String ownerUuid, NearCachingHook nearCachingHook) {
+        assert nearCachingHook != null;
+
         this.name = name;
         this.key = key;
         if (!(op instanceof MapTxnOperation)) {
@@ -51,6 +57,7 @@ public MapTransactionLogRecord(String name, Data key, int partitionId, Operation
         this.op = op;
         this.ownerUuid = ownerUuid;
         this.partitionId = partitionId;
+        this.nearCachingHook = nearCachingHook;
     }
 
     @Override
@@ -69,6 +76,20 @@ public Operation newCommitOperation() {
         return op;
     }
 
+    @Override
+    public void onCommitSuccess() {
+        assert nearCachingHook != null;
+
+        nearCachingHook.onRemoteCallSuccess();
+    }
+
+    @Override
+    public void onCommitFailure() {
+        assert nearCachingHook != null;
+
+        nearCachingHook.onRemoteCallFailure();
+    }
+
     @Override
     public Operation newRollbackOperation() {
         TxnRollbackOperation operation = new TxnRollbackOperation(partitionId, name, key, ownerUuid);