From fcd872c6439661c23f32befdd18019054b5006fe Mon Sep 17 00:00:00 2001
From: Rob Austin <rob.austin@boundedbuffer.co.uk>
Date: Fri, 29 May 2015 19:09:54 +0100
Subject: [PATCH] HCOLL-329 fixing the bootstrapping timestamp issue

Conflicts:
	src/main/java/net/openhft/chronicle/map/ChannelProvider.java
	src/main/java/net/openhft/chronicle/map/Replica.java
	src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
	src/main/java/net/openhft/chronicle/map/SharedSegment.java
	src/main/java/net/openhft/chronicle/map/TcpReplicator.java
	src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
---
 .../map/AbstractChannelReplicator.java        | 10 ++--
 .../chronicle/map/ChannelProvider.java        | 39 ++++++++++----
 .../net/openhft/chronicle/map/Replica.java    | 15 ++++--
 .../chronicle/map/ReplicatedChronicleMap.java | 51 +++++++++++++++----
 .../openhft/chronicle/map/TcpReplicator.java  |  4 +-
 .../map/impl/stage/input/ReplicatedInput.java |  9 ++--
 .../stage/replication/ReplicationUpdate.java  |  4 +-
 ...cketReplication3wayPutReturnsNullTest.java | 14 +++--
 .../chronicle/map/TestReplication.java        |  7 ++-
 9 files changed, 110 insertions(+), 43 deletions(-)

diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
index 980caca29..397e6d7e5 100644
--- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
@@ -484,7 +484,6 @@ abstract class AbstractChannelReplicator implements Closeable {
             out = result;
 
             assert out.capacity() == in.capacity();
-
             assert out.capacity() == size;
             assert out.capacity() == in.capacity();
             assert in.limit() == in.capacity();
@@ -498,7 +497,9 @@ abstract class AbstractChannelReplicator implements Closeable {
 
 
         @Override
-        public boolean onEntry(final Bytes entry, final int chronicleId) {
+        public boolean onEntry(@NotNull final Bytes entry,
+                               final int chronicleId,
+                               final long bootstrapTime) {
 
             long startOfEntry = entry.position();
             long pos0 = in.position();
@@ -514,7 +515,7 @@ abstract class AbstractChannelReplicator implements Closeable {
 
                 start = in.position();
 
-                externalizable.writeExternalEntry(entry, in, chronicleId);
+                externalizable.writeExternalEntry(entry, in, chronicleId, bootstrapTime);
 
                 if (in.position() == start) {
                     in.position(pos0);
@@ -558,7 +559,8 @@ abstract class AbstractChannelReplicator implements Closeable {
 
                     in.position(pos0);
                     entry.position(startOfEntry);
-                    return onEntry(entry, chronicleId);
+
+                    return onEntry(entry, chronicleId, bootstrapTime);
                 } else
                     throw e;
 
diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
index 126233fac..c4765efef 100644
--- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
@@ -18,7 +18,10 @@
 
 package net.openhft.chronicle.map;
 
-import net.openhft.chronicle.hash.replication.*;
+import net.openhft.chronicle.hash.replication.ReplicableEntry;
+import net.openhft.chronicle.hash.replication.ReplicationHub;
+import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
+import net.openhft.chronicle.hash.replication.UdpTransportConfig;
 import net.openhft.lang.collection.DirectBitSet;
 import net.openhft.lang.collection.SingleThreadedDirectBitSet;
 import net.openhft.lang.io.ByteBufferBytes;
@@ -111,7 +114,6 @@ public final class ChannelProvider implements Closeable {
 
         /**
          * writes the entry to the chronicle channel provided
-         *
          * @param entry            the byte location of the entry to be stored
          * @param destination      a buffer the entry will be written to, the segment may reject
          *                         this operation and add zeroBytes, if the identifier in the entry
@@ -119,21 +121,21 @@ public final class ChannelProvider implements Closeable {
          * @param chronicleChannel used in cluster into identify the canonical map or queue
          */
         @Override
-        public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
-                                       int chronicleChannel) {
+        public void writeExternalEntry(@NotNull Bytes entry,
+                                       @NotNull Bytes destination,
+                                       int chronicleChannel, long bootstrapTime) {
             channelDataReadLock();
             try {
                 destination.writeStopBit(chronicleChannel);
                 channelEntryExternalizables[chronicleChannel]
-                        .writeExternalEntry(entry, destination, chronicleChannel);
+                        .writeExternalEntry(entry, destination, chronicleChannel, bootstrapTime);
             } finally {
                 channelDataLock.readLock().unlock();
             }
         }
 
         @Override
-        public void readExternalEntry(
-                @NotNull Bytes source) {
+        public void readExternalEntry(@NotNull Bytes source) {
             channelDataReadLock();
             try {
                 final int chronicleId = (int) source.readStopBit();
@@ -263,6 +265,20 @@ public final class ChannelProvider implements Closeable {
             }
         }
 
+        @Override
+        public void setLastModificationTime(byte identifier, long timestamp) {
+            channelDataReadLock();
+            try {
+                // not including the SystemQueue at index 0
+                for (Replica channel : chronicleChannelMap.values()) {
+                    channel.setLastModificationTime(identifier, timestamp);
+                }
+
+            } finally {
+                channelDataLock.readLock().unlock();
+            }
+        }
+
         @Override
         public void close() throws IOException {
             ChannelProvider.this.close();
@@ -483,7 +499,7 @@ public final class ChannelProvider implements Closeable {
                         final Bytes bytes = payloads.poll();
                         if (bytes == null)
                             return false;
-                        callback.onEntry(bytes, 0);
+                        callback.onEntry(bytes, 0, System.currentTimeMillis());
                         return true;
                     }
 
@@ -519,6 +535,11 @@ public final class ChannelProvider implements Closeable {
                 return 0;
             }
 
+            @Override
+            public void setLastModificationTime(byte identifier, long timestamp) {
+                // do nothing
+            }
+
             @Override
             public void close() throws IOException {
                 // do nothing
@@ -537,7 +558,7 @@ public final class ChannelProvider implements Closeable {
 
             @Override
             public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
-                                           int na) {
+                                           int na, long bootstrapTime) {
                 destination.write(entry, entry.position(), entry.remaining());
             }
 
diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java
index 9c504bb1f..7dcce3b24 100644
--- a/src/main/java/net/openhft/chronicle/map/Replica.java
+++ b/src/main/java/net/openhft/chronicle/map/Replica.java
@@ -69,6 +69,8 @@ public interface Replica extends Closeable {
      */
     long lastModificationTime(byte remoteIdentifier);
 
+    void setLastModificationTime(byte identifier, long timestamp);
+
     /**
      * notifies when there is a changed to the modification iterator
      */
@@ -156,8 +158,8 @@ public interface Replica extends Closeable {
          *                    match the maps local
          * @param chronicleId is the channel id used to identify the canonical map or queue
          */
-        void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination, int
-                chronicleId);
+        void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
+                                int chronicleId, long bootstrapTime);
 
         /**
          * The map implements this method to restore its contents. This method must read the values
@@ -188,16 +190,19 @@ public interface Replica extends Closeable {
          * source node is not from one of our changes, WARNING even though we check the identifier
          * in the ModificationIterator the entry may have been updated.
          */
-        public abstract boolean onEntry(final Bytes entry, final int chronicleId);
+        public abstract boolean onEntry(
+                final Bytes entry, final int chronicleId, long bootstrapTime);
 
         /**
-         * Called just after {@link  #onEntry(net.openhft.lang.io.Bytes, int)}. No-op by default.
+         * Called just after {@link  #onEntry(net.openhft.lang.io.Bytes, int, long)}.
+         * No-op by default.
          */
         public void onAfterEntry() {
         }
 
         /**
-         * Called just before {@link #onEntry(net.openhft.lang.io.Bytes, int)}. No-op by default.
+         * Called just before {@link #onEntry(net.openhft.lang.io.Bytes, int, long)}.
+         * No-op by default.
          */
         public void onBeforeEntry() {
         }
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
index b5589e21d..1522e0b6f 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
@@ -39,6 +39,8 @@ import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReferenceArray;
 
 import static net.openhft.lang.MemoryUnit.*;
@@ -92,6 +94,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
     private static final long serialVersionUID = 0L;
     private static final Logger LOG = LoggerFactory.getLogger(ReplicatedChronicleMap.class);
     private static final long LAST_UPDATED_HEADER_SIZE = 128L * 8L;
+    private static final int SIZE_OF_BOOTSTRAP_TIME_STAMP = 8;
+
     public final TimeProvider timeProvider;
     private final byte localIdentifier;
     transient Set<Closeable> closeables;
@@ -266,11 +270,11 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
         }
     }
 
-    public void raiseChange(long segmentIndex, long pos) {
+    public void raiseChange(long segmentIndex, long pos, long timestamp) {
         for (long next = modIterSet.nextSetBit(0L); next > 0L;
              next = modIterSet.nextSetBit(next + 1L)) {
             try {
-                acquireModificationIterator((byte) next).raiseChange(segmentIndex, pos);
+                acquireModificationIterator((byte) next).raiseChange(segmentIndex, pos, timestamp);
             } catch (Exception e) {
                 LOG.error("", e);
             }
@@ -288,14 +292,14 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
         }
     }
 
-    public void moveChange(long segmentIndex, long oldPos, long newPos) {
+    public void moveChange(long segmentIndex, long oldPos, long newPos, long timestamp) {
         for (long next = modIterSet.nextSetBit(0L); next > 0L;
              next = modIterSet.nextSetBit(next + 1L)) {
             try {
                 ModificationIterator modificationIterator =
                         acquireModificationIterator((byte) next);
                 if (modificationIterator.dropChange(segmentIndex, oldPos))
-                    modificationIterator.raiseChange(segmentIndex, newPos);
+                    modificationIterator.raiseChange(segmentIndex, newPos, timestamp);
             } catch (Exception e) {
                 LOG.error("", e);
             }
@@ -332,7 +336,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
             // make a byte buffer
             assert result < Integer.MAX_VALUE;
 
-            return (int) result;
+            return (int) result + SIZE_OF_BOOTSTRAP_TIME_STAMP;
         } finally {
             entry.position(start);
         }
@@ -346,7 +350,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
     @Override
     public void writeExternalEntry(@NotNull Bytes entry,
                                    @NotNull Bytes destination,
-                                   int chronicleId) {
+                                   int chronicleId,
+                                   long bootstrapTime) {
 
         final long keySize = keySizeMarshaller.readSize(entry);
 
@@ -369,7 +374,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
         }
 
         final long valuePosition = entry.position();
-
+        destination.writeLong(bootstrapTime);
         keySizeMarshaller.writeSize(destination, keySize);
         valueSizeMarshaller.writeSize(destination, valueSize);
         destination.writeStopBit(timeStamp);
@@ -492,6 +497,12 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
         private final int segmentIndexShift;
         private final long posMask;
 
+        // when a bootstrap is send this is the time stamp that the client will bootstrap up to
+        // if it is set as ZERO then the onPut() will set it to the current time, once the
+        // consumer has cycled through the bit set the timestamp will be set back to zero.
+        private AtomicLong bootStrapTimeStamp = new AtomicLong();
+        private long lastBootStrapTimeStamp = timeProvider.currentTime();
+
         // records the current position of the cursor in the bitset
         private volatile long position = -1L;
 
@@ -524,10 +535,15 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
             return (segmentIndex << segmentIndexShift) | pos;
         }
 
-        void raiseChange(long segmentIndex, long pos) {
+        void raiseChange(long segmentIndex, long pos, long timestamp) {
             LOG.debug("raise change: id {}, segment {}, pos {}",
                     localIdentifier, segmentIndex, pos);
             changesForUpdates.set(combine(segmentIndex, pos));
+            assert timestamp > timeProvider.currentTime() - TimeUnit.SECONDS.toMillis(1) &&
+                    timestamp <= timeProvider.currentTime() : "timeStamp=" + timestamp + ", " +
+                    "currentTime=" + timeProvider.currentTime();
+            // todo improve this - use the timestamp from the entry its self
+            bootStrapTimeStamp.compareAndSet(0, timestamp);
             if (modificationNotifier != null)
                 modificationNotifier.onChange();
         }
@@ -568,6 +584,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
                         this.position = NOT_FOUND;
                         return false;
                     }
+
+                    bootStrapTimeStamp.set(0);
                     continue;
                 }
 
@@ -594,7 +612,8 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
                         // process it later, by NOT clearing the changes.clear(position)
                         context.entryBytes().limit(context.valueOffset() + context.valueSize());
                         context.entryBytes().position(context.keySizeOffset());
-                        boolean success = entryCallback.onEntry(context.entryBytes(), chronicleId);
+                        boolean success = entryCallback.onEntry(
+                                context.entryBytes(), chronicleId, bootStrapTimeStamp());
                         entryCallback.onAfterEntry();
 
                         if (success)
@@ -609,6 +628,18 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
             }
         }
 
+        /**
+         * @return the timestamp  that the remote client should bootstrap from when there has been a
+         * disconnection, this time maybe later than the message time as event are not send in
+         * chronological order from the bit set.
+         */
+        private long bootStrapTimeStamp() {
+            final long timeStamp = bootStrapTimeStamp.get();
+            long result = (timeStamp == 0) ? this.lastBootStrapTimeStamp : timeStamp;
+            this.lastBootStrapTimeStamp = result;
+            return result;
+        }
+
         @Override
         public void dirtyEntries(long fromTimeStamp) {
             try (CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R, ?> c =
@@ -636,7 +667,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
                         if (re.originTimestamp() >= fromTimeStamp &&
                                 (!bootstrapOnlyLocalEntries ||
                                         re.originIdentifier() == localIdentifier)) {
-                            raiseChange(segmentIndex, c.pos());
+                            raiseChange(segmentIndex, c.pos(), c.timestamp());
                         }
                     });
                 }
diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
index 103cb7f7a..8cab4d7b8 100644
--- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
@@ -1282,6 +1282,7 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
      */
     class TcpSocketChannelEntryReader {
         public static final int HEADROOM = 1024;
+        public static final int SIZE_OF_BOOTSTRAP_TIMESTAMP = 8;
         public long lastHeartBeatReceived = System.currentTimeMillis();
         ByteBuffer in;
         ByteBufferBytes out;
@@ -1367,7 +1368,8 @@ public final class TcpReplicator<K, V> extends AbstractChannelReplicator impleme
 
                         // if the buffer is too small to read this payload we will have to grow the
                         // size of the buffer
-                        long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1;
+                        long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1 +
+                                SIZE_OF_BOOTSTRAP_TIMESTAMP;
                         if (out.capacity() < requiredSize) {
                             attached.entryReader.resizeBuffer(requiredSize + HEADROOM);
                         }
diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java b/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java
index 68ad3bac7..3cd15f4c3 100644
--- a/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java
+++ b/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java
@@ -46,7 +46,8 @@ public abstract class ReplicatedInput<K, V, R>
         this.replicatedInputBytes = replicatedInputBytes;
         replicatedInputStore.setBytes(replicatedInputBytes);
     }
-    
+
+    @Stage("ReplicationInput") public long bootstrapTimestamp;
     @Stage("ReplicationInput") public long riKeySize = -1;
     @Stage("ReplicationInput") public long riValueSize;
 
@@ -60,6 +61,7 @@ public abstract class ReplicatedInput<K, V, R>
 
     public void initReplicationInput(Bytes replicatedInputBytes) {
         initReplicatedInputBytes(replicatedInputBytes);
+        bootstrapTimestamp = replicatedInputBytes.readLong();
         riKeySize = mh.m().keySizeMarshaller.readSize(replicatedInputBytes);
         riValueSize = mh.m().valueSizeMarshaller.readSize(replicatedInputBytes);
 
@@ -79,6 +81,8 @@ public abstract class ReplicatedInput<K, V, R>
             return;
         }
 
+        mh.m().setLastModificationTime(riId, bootstrapTimestamp);
+
         q.initInputKey(replicatedInputKeyBytesValue);
 
         boolean debugEnabled = lh.LOG.isDebugEnabled();
@@ -90,7 +94,6 @@ public abstract class ReplicatedInput<K, V, R>
                         mh.m().identifier(), riId, q.inputKey);
             }
             mh.m().remoteOperations.remove(this);
-            mh.m().setLastModificationTime(riId, riTimestamp);
             return;
         }
 
@@ -103,7 +106,7 @@ public abstract class ReplicatedInput<K, V, R>
 
 
         mh.m().remoteOperations.put(this, replicatedInputValueBytesValue);
-        mh.m().setLastModificationTime(riId, riTimestamp);
+
 
         if (debugEnabled) {
             lh.LOG.debug(message + "value=" + replicatedInputValueBytesValue + ")");
diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java
index 8c596daac..0f07fe84c 100644
--- a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java
+++ b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java
@@ -51,7 +51,7 @@ public abstract class ReplicationUpdate<K> implements RemoteOperationContext<K>
     }
 
     public void moveChange(long oldPos, long newPos) {
-        mh.m().moveChange(s.segmentIndex, oldPos, newPos);
+        mh.m().moveChange(s.segmentIndex, oldPos, newPos, e.timestamp());
     }
     
     public void updateChange() {
@@ -61,7 +61,7 @@ public abstract class ReplicationUpdate<K> implements RemoteOperationContext<K>
     }
 
     public void raiseChange() {
-        mh.m().raiseChange(s.segmentIndex, e.pos);
+        mh.m().raiseChange(s.segmentIndex, e.pos, e.timestamp());
     }
 
     @Override
diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication3wayPutReturnsNullTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication3wayPutReturnsNullTest.java
index 975ff78f5..63bdc0e24 100755
--- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication3wayPutReturnsNullTest.java
+++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication3wayPutReturnsNullTest.java
@@ -21,7 +21,6 @@ package net.openhft.chronicle.map;
 import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
 import org.junit.After;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import java.io.Closeable;
@@ -56,12 +55,17 @@ public class TCPSocketReplication3wayPutReturnsNullTest {
                 .replication(identifier, tcpConfig).create();
     }
 
+    int portOffset = 0;
+
     @Before
     public void setup() throws IOException {
-        map1 = newTcpSocketShmIntString((byte) 1, 8026, new InetSocketAddress("localhost", 8027),
-                new InetSocketAddress("localhost", 8028));
-        map2 = newTcpSocketShmIntString((byte) 2, 8027, new InetSocketAddress("localhost", 8028));
-        map3 = newTcpSocketShmIntString((byte) 3, 8028);
+        map1 = newTcpSocketShmIntString((byte) 1, portOffset + 8026, new InetSocketAddress
+                        ("localhost", portOffset + 8027),
+                new InetSocketAddress("localhost", portOffset + 8028));
+        map2 = newTcpSocketShmIntString((byte) 2, portOffset + 8027, new InetSocketAddress
+                ("localhost", portOffset + 8028));
+        map3 = newTcpSocketShmIntString((byte) 3, portOffset + 8028);
+        portOffset += 10;
     }
 
     @After
diff --git a/src/test/java/net/openhft/chronicle/map/TestReplication.java b/src/test/java/net/openhft/chronicle/map/TestReplication.java
index cea4911c3..83b2cc7cf 100644
--- a/src/test/java/net/openhft/chronicle/map/TestReplication.java
+++ b/src/test/java/net/openhft/chronicle/map/TestReplication.java
@@ -18,11 +18,11 @@ public class TestReplication {
 
 
         TcpTransportAndNetworkConfig tcpConfigServer1 =
-                TcpTransportAndNetworkConfig.of(8082);
+                TcpTransportAndNetworkConfig.of(8092);
 
         TcpTransportAndNetworkConfig tcpConfigServer2 =
-                TcpTransportAndNetworkConfig.of(8083, new InetSocketAddress("localhost",
-                        8082));
+                TcpTransportAndNetworkConfig.of(8093, new InetSocketAddress("localhost",
+                        8092));
 
         final ChronicleMap<Integer, Integer> map2 = ChronicleMapBuilder.of(Integer.class,
                 Integer.class)
@@ -40,7 +40,6 @@ public class TestReplication {
         }
 
         for (int i = 0; i < 10; i++) {
-
             Thread.sleep(100);
             System.out.println(map2.size());
         }