diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
index 390337a06..94c8ab798 100644
--- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
@@ -477,6 +477,10 @@ abstract class AbstractChannelReplicator implements Closeable {
 
         }
 
+        public boolean shouldBeIgnored(final Bytes entry, final int chronicleId) {
+            return !externalizable.identifierCheck(entry, chronicleId);
+        }
+
 
         @Override
         public boolean onEntry(final Bytes entry, final int chronicleId) {
@@ -494,8 +498,14 @@ abstract class AbstractChannelReplicator implements Closeable {
             int entrySize = externalizable.sizeOfEntry(entry, chronicleId);
 
             if (entrySize > in.remaining()) {
+
                 long newSize = in.position() + entrySize;
-                assert newSize < Integer.MAX_VALUE;
+
+                // This can occur when we pack a number of entries into the buffer and the
+                // last entry is very large.
+                if (newSize > Integer.MAX_VALUE)
+                    return false;
+
                 resizeBuffer((int) newSize);
             }
 
diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
index addfb502e..b69f8c3bb 100644
--- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
@@ -97,6 +97,18 @@ final class ChannelProvider implements Closeable {
             }
         }
 
+        @Override
+        public boolean identifierCheck(@NotNull Bytes entry, int chronicleChannel) {
+            channelDataLock.readLock().lock();
+            try {
+
+                return channelEntryExternalizables[chronicleChannel]
+                        .identifierCheck(entry, chronicleChannel);
+            } finally {
+                channelDataLock.readLock().unlock();
+            }
+        }
+
         /**
          * writes the entry to the chronicle channel provided
          *
@@ -484,6 +496,11 @@ final class ChannelProvider implements Closeable {
                 return (int) entry.remaining();
             }
 
+            @Override
+            public boolean identifierCheck(@NotNull Bytes entry, int chronicleId) {
+                return true;
+            }
+
             @Override
             public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
                                            int na) {
diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java
index 9f328e74f..1077c44f7 100644
--- a/src/main/java/net/openhft/chronicle/map/Replica.java
+++ b/src/main/java/net/openhft/chronicle/map/Replica.java
@@ -128,12 +128,25 @@ interface Replica extends Closeable {
     interface EntryExternalizable {
 
         /**
-         * @param entry an entry in the map
-         * @param chronicleId  is the channel id used to identify the canonical map or queue
+         * The size of entry in bytes
+         *
+         * @param entry       an entry in the map
+         * @param chronicleId is the channel id used to identify the canonical map or queue
          * @return the size of the entry
          */
         int sizeOfEntry(@NotNull Bytes entry, int chronicleId);
 
+
+        /**
+         * check that the identifier in the entry is from this node
+         *
+         * @param entry       an entry in the map
+         * @param chronicleId is the channel id used to identify the canonical map or queue
+         * @return the size of the entry
+         */
+        boolean identifierCheck(@NotNull Bytes entry, int chronicleId);
+
+
         /**
          * The map implements this method to save its contents.
          *
@@ -160,6 +173,8 @@ interface Replica extends Closeable {
         void readExternalEntry(
                 @NotNull ThreadLocalCopies copies,
                 @NotNull VanillaChronicleMap.SegmentState segmentState, @NotNull Bytes source);
+
+
     }
 
     /**
@@ -218,6 +233,21 @@ interface Replica extends Closeable {
          */
         public void onBeforeEntry() {
         }
+
+
+        /**
+         * its possible that the entry should now be ignored, for example although rare its
+         * identifier may have recently been changed by another thread, so its no longer applicable
+         * to send
+         *
+         * @param entry       the entry you will receive, this does not have to be locked, as
+         *                    locking is already provided from the caller.
+         * @param chronicleId only assigned when clustering
+         * @return {@code true} if this entry should be ignored
+         */
+        public abstract boolean shouldBeIgnored(final Bytes entry, final int chronicleId);
     }
+
+
 }
 
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
index b70aa56dd..43fb00602 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
@@ -422,31 +422,53 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
         }
     }
 
+    public boolean identifierCheck(@NotNull Bytes entry, int chronicleId) {
+        long start = entry.position();
+        try {
+            final long keySize = keySizeMarshaller.readSize(entry);
+            entry.skip(keySize + 8); // we skip 8 for the timestamp
+            final byte identifier = entry.readByte();
+            return identifier == localIdentifier;
+        } finally {
+            entry.position(start);
+        }
+    }
+
+
     public int sizeOfEntry(@NotNull Bytes entry, int chronicleId) {
 
         long start = entry.position();
-        final long keySize = keySizeMarshaller.readSize(entry);
+        try {
+            final long keySize = keySizeMarshaller.readSize(entry);
 
-        entry.skip(keySize + ADDITIONAL_ENTRY_BYTES - 1L);
+            entry.skip(keySize + 8); // we skip 8 for the timestamp
 
-        final boolean isDeleted = entry.readBoolean();
-        long valueSize;
-        if (!isDeleted) {
-            valueSize = valueSizeMarshaller.readSize(entry);
-        } else {
-            valueSize = 0L;
+            final byte identifier = entry.readByte();
+            if (identifier != localIdentifier) {
+                // although unlikely, this may occur if the entry has been updated
+                return 0;
+            }
+
+            final boolean isDeleted = entry.readBoolean();
+            long valueSize;
+            if (!isDeleted) {
+                valueSize = valueSizeMarshaller.readSize(entry);
+            } else {
+                valueSize = 0L;
+            }
+
+            alignment.alignPositionAddr(entry);
+            long result = (entry.position() + valueSize - start);
+
+            // entries can be larger than Integer.MAX_VALUE as we are restricted to the size we can
+            // make a byte buffer
+            assert result < Integer.MAX_VALUE;
+
+            return (int) result;
+        } finally {
+            entry.position(start);
         }
 
-        alignment.alignPositionAddr(entry);
-        long result = (entry.position() + valueSize - start);
-        entry.position(start);
-
-        // entries can be larger than Integer.MAX_VALUE as we are restricted to the size we can
-        // make a byte buffer
-        assert result < Integer.MAX_VALUE;
-
-        return (int) result;
-
     }
 
 
@@ -455,7 +477,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
      * this method, especially when being used in a multi threaded context.
      */
     @Override
-    public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
+    public void writeExternalEntry(@NotNull Bytes entry,
+                                   @NotNull Bytes destination,
                                    int chronicleId) {
         final long initialLimit = entry.limit();
 
@@ -1514,7 +1537,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
                         segment((int) (position >>> segmentIndexShift));
                 segment.readLock(null);
                 try {
-                    if (changes.clearIfSet(position)) {
+                    if (changes.get(position)) {
+
                         entryCallback.onBeforeEntry();
 
                         final long segmentPos = position & posMask;
@@ -1522,13 +1546,24 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
                                 segment.reuse(tmpBytes, segment.offsetFromPos(segmentPos));
 
                         // if the entry should be ignored, we'll move the next entry
+                        if (entryCallback.shouldBeIgnored(entry, chronicleId)) {
+                            changes.clear(position);
+                            continue;
+                        }
+
+                        // it may not be successful if the buffer can not be resided so we will
+                        // process it later, by NOT clearing the changes.clear(position)
                         final boolean success = entryCallback.onEntry(entry, chronicleId);
                         entryCallback.onAfterEntry();
-                        if (success) {
-                            return true;
-                        }
+
+                        if (success)
+                            changes.clear(position);
+
+                        return success;
+
                     }
 
+
                     // if the position was already cleared by another thread
                     // while we were trying to obtain segment lock (for example, in onRelocation()),
                     // go to pick up next (next iteration in while (true) loop)
diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
index b09c2effe..66bb147d4 100644
--- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
@@ -593,13 +593,18 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
             socketChannel.close();
             return;
         }
+
         TcpSocketChannelEntryWriter entryWriter = attached.entryWriter;
-        if (entryWriter == null) throw new NullPointerException("No entryWriter");
+
+        if (entryWriter == null) throw
+                new NullPointerException("No entryWriter");
+
         if (entryWriter.isWorkIncomplete()) {
             final boolean completed = entryWriter.doWork();
 
             if (completed)
                 entryWriter.workCompleted();
+
         } else if (attached.remoteModificationIterator != null)
             entryWriter.entriesToBuffer(attached.remoteModificationIterator, key);
 
@@ -986,25 +991,34 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
                 for (; ; entriesWritten++) {
 
                     long start = in().position();
-                    final boolean isMoreDataAvailable = modificationIterator.nextEntry(entryCallback, 0);
+
+                    boolean success = modificationIterator.nextEntry(entryCallback, 0);
+
+                    // if not success this is most likely due to the next entry not fitting into
+                    // the buffer and the buffer can not be re-sized above Integer.max_value, in
+                    // this case success will return false, so we return so that we can send to
+                    // the socket what we have.
+                    if (!success)
+                        return;
+
                     long entrySize = in().position() - start;
 
                     if (entrySize > largestEntrySoFar)
                         largestEntrySoFar = entrySize;
 
 
-                    if (!isMoreDataAvailable) {
+                    if (!modificationIterator.hasNext()) {
                         // if we have no more data to write to the socket then we will
                         // un-register OP_WRITE on the selector, until more data becomes available
                         if (in().position() == 0 && handShakingComplete)
                             disableWrite(socketChannel, attached);
 
-                        return;
                     }
 
                     // we've filled up the buffer lets give another channel a chance to send
                     // some data
-                    if (in().remaining() <= largestEntrySoFar)
+                    if (in().remaining() <= largestEntrySoFar || in().position() >
+                            replicationConfig.tcpBufferSize())
                         return;
 
                     // if we have space in the buffer to write more data and we just wrote data
@@ -1027,7 +1041,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
                                         final long approxTime) throws IOException {
 
             final Bytes in = in();
-            final  ByteBuffer out = out();
+            final ByteBuffer out = out();
 
             if (in.position() == 0)
                 return 0;