Add ReplicableEntry.dropChangeFor(remoteIdentifier)

src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java

@@ -42,6 +42,11 @@ public interface ReplicableEntryDelegating extends ReplicableEntry {
         d().dropChanged();
     }
 
+    @Override
+    default void dropChangedFor(byte remoteIdentifier) {
+        d().dropChangedFor(remoteIdentifier);
+    }
+
     @Override
     default void raiseChanged() {
         d().raiseChanged();

src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java

@@ -76,13 +76,19 @@ public interface ReplicableEntry {
      */
     void dropChanged();
 
+    /**
+     * Suppress the entry, if it was scheduled to be replicated over to the remote Chronicle node
+     * with the specified identifier.
+     */
+    void dropChangedFor(byte remoteIdentifier);
+
     /**
      * Propagate the entry, schedule it to be replicated over to all remote Chronicle nodes.
      */
     void raiseChanged();
 
     /**
-     * Propagate the entry, schedule it to be replicated over to the remote Chronicle nodes with
+     * Propagate the entry, schedule it to be replicated over to the remote Chronicle node with
      * the specified identifier.
      *
      * @param remoteIdentifier the identifier of the node to replicate this entry to

src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java

@@ -380,6 +380,10 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
         }
     }
 
+    public void dropChangeFor(long tierIndex, long pos, byte remoteIdentifier) {
+        acquireModificationIterator(remoteIdentifier).dropChange0(tierIndex, pos);
+    }
+
     public void moveChange(long oldTierIndex, long oldPos, long newTierIndex, long newPos) {
         if (oldTierIndex <= actualSegments) {
             long oldSegmentIndex = oldTierIndex - 1;
@@ -899,5 +903,19 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
                 raiseChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos);
             }
         }
+
+        void dropChange0(long tierIndex, long pos) {
+            if (tierIndex <= actualSegments) {
+                long segmentIndex = tierIndex - 1;
+                long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize;
+                dropChangeInSegment(offsetToTierBitSet, pos);
+            } else {
+                long extraTierIndex = tierIndex - 1 - actualSegments;
+                int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk);
+                long offsetToTierBitSet =
+                        (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize;
+                dropChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos);
+            }
+        }
     }
 }

src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java

@@ -112,6 +112,13 @@ public abstract class ReplicatedMapEntryStages<K, V> extends MapEntryStages<K, V
         ru.dropChange();
     }
 
+    @Override
+    public void dropChangedFor(byte remoteIdentifier) {
+        checkOnEachPublicOperation.checkOnEachPublicOperation();
+        s.innerUpdateLock.lock();
+        ru.dropChangeFor(remoteIdentifier);
+    }
+
     @Override
     public void raiseChanged() {
         checkOnEachPublicOperation.checkOnEachPublicOperation();

src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java

@@ -52,6 +52,10 @@ public abstract class ReplicationUpdate<K> implements RemoteOperationContext<K>
         mh.m().dropChange(s.tierIndex, e.pos);
     }
 
+    public void dropChangeFor(byte remoteIdentifier) {
+        mh.m().dropChangeFor(s.tierIndex, e.pos, remoteIdentifier);
+    }
+
     public void moveChange(long oldTierIndex, long oldPos, long newPos) {
         mh.m().moveChange(oldTierIndex, oldPos, s.tierIndex, newPos);
     }