diff --git a/src/main/java/org/apache/gossip/examples/StandAloneNode.java b/src/main/java/org/apache/gossip/examples/StandAloneNode.java
index c12f946..d24c0fa 100644
--- a/src/main/java/org/apache/gossip/examples/StandAloneNode.java
+++ b/src/main/java/org/apache/gossip/examples/StandAloneNode.java
@@ -31,7 +31,7 @@
     GossipSettings s = new GossipSettings();
     s.setWindowSize(10);
     s.setConvictThreshold(1.0);
-    s.setGossipInterval(10);
+    s.setGossipInterval(1000);
     GossipService gossipService = new GossipService("mycluster",  URI.create(args[0]), args[1],
             Arrays.asList( new RemoteGossipMember("mycluster", URI.create(args[2]), args[3])), s, (a,b) -> {}, new MetricRegistry());
     gossipService.start();
diff --git a/src/main/java/org/apache/gossip/manager/ActiveGossipThread.java b/src/main/java/org/apache/gossip/manager/ActiveGossipThread.java
index 731b019..f81565b 100644
--- a/src/main/java/org/apache/gossip/manager/ActiveGossipThread.java
+++ b/src/main/java/org/apache/gossip/manager/ActiveGossipThread.java
@@ -17,7 +17,6 @@
  */
 package org.apache.gossip.manager;
 
-import java.io.IOException;
 import java.util.List;
 
 import java.util.Map.Entry;
@@ -44,8 +43,6 @@
 import org.apache.gossip.udp.UdpSharedGossipDataMessage;
 import org.apache.log4j.Logger;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-
 import static com.codahale.metrics.MetricRegistry.name;
 
 /**
@@ -61,7 +58,6 @@
   private ScheduledExecutorService scheduledExecutorService;
   private final BlockingQueue<Runnable> workQueue;
   private ThreadPoolExecutor threadService;
-  private ObjectMapper MAPPER = new ObjectMapper();
 
   private final Histogram sharedDataHistogram;
   private final Histogram sendPerNodeDataHistogram;
@@ -114,28 +110,17 @@
       LOGGER.debug("Send sendMembershipList() is called without action");
       sharedDataHistogram.update(System.currentTimeMillis() - startTime);
       return;
-    }
-    try {
-      for (Entry<String, SharedGossipDataMessage> innerEntry : this.gossipCore.getSharedData().entrySet()){
-          UdpSharedGossipDataMessage message = new UdpSharedGossipDataMessage();
-          message.setUuid(UUID.randomUUID().toString());
-          message.setUriFrom(me.getId());
-          message.setExpireAt(innerEntry.getValue().getExpireAt());
-          message.setKey(innerEntry.getValue().getKey());
-          message.setNodeId(innerEntry.getValue().getNodeId());
-          message.setTimestamp(innerEntry.getValue().getTimestamp());
-          message.setPayload(innerEntry.getValue().getPayload());
-          byte[] json_bytes = MAPPER.writeValueAsString(message).getBytes();
-          int packet_length = json_bytes.length;
-          if (packet_length < GossipManager.MAX_PACKET_SIZE) {
-            gossipCore.sendOneWay(message, member.getUri());
-          } else {
-            LOGGER.error("The length of the to be send message is too large ("
-                    + packet_length + " > " + GossipManager.MAX_PACKET_SIZE + ").");
-          }
-      }
-    } catch (IOException e1) {
-      LOGGER.warn(e1);
+    }    
+    for (Entry<String, SharedGossipDataMessage> innerEntry : gossipCore.getSharedData().entrySet()){
+        UdpSharedGossipDataMessage message = new UdpSharedGossipDataMessage();
+        message.setUuid(UUID.randomUUID().toString());
+        message.setUriFrom(me.getId());
+        message.setExpireAt(innerEntry.getValue().getExpireAt());
+        message.setKey(innerEntry.getValue().getKey());
+        message.setNodeId(innerEntry.getValue().getNodeId());
+        message.setTimestamp(innerEntry.getValue().getTimestamp());
+        message.setPayload(innerEntry.getValue().getPayload());
+        gossipCore.sendOneWay(message, member.getUri());
     }
     sharedDataHistogram.update(System.currentTimeMillis() - startTime);
   }
@@ -148,36 +133,26 @@
       LOGGER.debug("Send sendMembershipList() is called without action");
       sendPerNodeDataHistogram.update(System.currentTimeMillis() - startTime);
       return;
-    }
-    try {
-      for (Entry<String, ConcurrentHashMap<String, GossipDataMessage>> entry : gossipCore.getPerNodeData().entrySet()){
-        for (Entry<String, GossipDataMessage> innerEntry : entry.getValue().entrySet()){
-          UdpGossipDataMessage message = new UdpGossipDataMessage();
-          message.setUuid(UUID.randomUUID().toString());
-          message.setUriFrom(me.getId());
-          message.setExpireAt(innerEntry.getValue().getExpireAt());
-          message.setKey(innerEntry.getValue().getKey());
-          message.setNodeId(innerEntry.getValue().getNodeId());
-          message.setTimestamp(innerEntry.getValue().getTimestamp());
-          message.setPayload(innerEntry.getValue().getPayload());
-          byte[] json_bytes = MAPPER.writeValueAsString(message).getBytes();
-          int packet_length = json_bytes.length;
-          if (packet_length < GossipManager.MAX_PACKET_SIZE) {
-            gossipCore.sendOneWay(message, member.getUri());
-          } else {
-            LOGGER.error("The length of the to be send message is too large ("
-                    + packet_length + " > " + GossipManager.MAX_PACKET_SIZE + ").");
-          }
-        }
+    }    
+    for (Entry<String, ConcurrentHashMap<String, GossipDataMessage>> entry : gossipCore.getPerNodeData().entrySet()){
+      for (Entry<String, GossipDataMessage> innerEntry : entry.getValue().entrySet()){
+        UdpGossipDataMessage message = new UdpGossipDataMessage();
+        message.setUuid(UUID.randomUUID().toString());
+        message.setUriFrom(me.getId());
+        message.setExpireAt(innerEntry.getValue().getExpireAt());
+        message.setKey(innerEntry.getValue().getKey());
+        message.setNodeId(innerEntry.getValue().getNodeId());
+        message.setTimestamp(innerEntry.getValue().getTimestamp());
+        message.setPayload(innerEntry.getValue().getPayload());
+        gossipCore.sendOneWay(message, member.getUri());   
       }
-    } catch (IOException e1) {
-      LOGGER.warn(e1);
     }
     sendPerNodeDataHistogram.update(System.currentTimeMillis() - startTime);
   }
   
   protected void sendToALiveMember(){
     LocalGossipMember member = selectPartner(gossipManager.getLiveMembers());
+    System.out.println("send" );
     sendMembershipList(gossipManager.getMyself(), member);
   }
   
@@ -199,29 +174,18 @@
     } else {
       LOGGER.debug("Send sendMembershipList() is called to " + member.toString());
     }
-    try {
-      UdpActiveGossipMessage message = new UdpActiveGossipMessage();
-      message.setUriFrom(gossipManager.getMyself().getUri().toASCIIString());
-      message.setUuid(UUID.randomUUID().toString());
-      message.getMembers().add(convert(me));
-      for (LocalGossipMember other : gossipManager.getMembers().keySet()) {
-        message.getMembers().add(convert(other));
-      }
-      byte[] json_bytes = MAPPER.writeValueAsString(message).getBytes();
-      int packet_length = json_bytes.length;
-      if (packet_length < GossipManager.MAX_PACKET_SIZE) {
-        Response r = gossipCore.send(message, member.getUri());
-        if (r instanceof ActiveGossipOk){
-          //maybe count metrics here
-        } else {
-          LOGGER.debug("Message " + message + " generated response " + r);
-        }
-      } else {
-        LOGGER.error("The length of the to be send message is too large ("
-                + packet_length + " > " + GossipManager.MAX_PACKET_SIZE + ").");
-      }
-    } catch (IOException e1) {
-      LOGGER.warn(e1);
+    UdpActiveGossipMessage message = new UdpActiveGossipMessage();
+    message.setUriFrom(gossipManager.getMyself().getUri().toASCIIString());
+    message.setUuid(UUID.randomUUID().toString());
+    message.getMembers().add(convert(me));
+    for (LocalGossipMember other : gossipManager.getMembers().keySet()) {
+      message.getMembers().add(convert(other));
+    }
+    Response r = gossipCore.send(message, member.getUri());
+    if (r instanceof ActiveGossipOk){
+      //maybe count metrics here
+    } else {
+      LOGGER.debug("Message " + message + " generated response " + r);
     }
     sendMembershipHistorgram.update(System.currentTimeMillis() - startTime);
   }
diff --git a/src/main/java/org/apache/gossip/manager/GossipCore.java b/src/main/java/org/apache/gossip/manager/GossipCore.java
index d315361..5d561c3 100644
--- a/src/main/java/org/apache/gossip/manager/GossipCore.java
+++ b/src/main/java/org/apache/gossip/manager/GossipCore.java
@@ -31,7 +31,6 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -53,27 +52,42 @@
 import org.apache.gossip.udp.UdpSharedGossipDataMessage;
 import org.apache.log4j.Logger;
 
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 
-public class GossipCore {
+public class GossipCore implements GossipCoreConstants {
   
   public static final Logger LOGGER = Logger.getLogger(GossipCore.class);
   private static final ObjectMapper MAPPER = new ObjectMapper();
   private final GossipManager gossipManager;
   private ConcurrentHashMap<String, Base> requests;
-  private ExecutorService service;
+  private ThreadPoolExecutor service;
   private final ConcurrentHashMap<String, ConcurrentHashMap<String, GossipDataMessage>> perNodeData;
   private final ConcurrentHashMap<String, SharedGossipDataMessage> sharedData;
   private final BlockingQueue<Runnable> workQueue;
+  private final Meter messageSerdeException;
+  private final Meter tranmissionException;
+  private final Meter tranmissionSuccess;
   
-  public GossipCore(GossipManager manager){
+  public GossipCore(GossipManager manager, MetricRegistry metrics){
     this.gossipManager = manager;
     requests = new ConcurrentHashMap<>();
     workQueue = new ArrayBlockingQueue<>(1024);
     service = new ThreadPoolExecutor(1, 5, 1, TimeUnit.SECONDS, workQueue, new ThreadPoolExecutor.DiscardOldestPolicy());
     perNodeData = new ConcurrentHashMap<>();
     sharedData = new ConcurrentHashMap<>();
+    metrics.register(WORKQUEUE_SIZE, (Gauge<Integer>)() -> workQueue.size());
+    metrics.register(PER_NODE_DATA_SIZE, (Gauge<Integer>)() -> perNodeData.size());
+    metrics.register(SHARED_DATA_SIZE, (Gauge<Integer>)() ->  sharedData.size());
+    metrics.register(REQUEST_SIZE, (Gauge<Integer>)() ->  requests.size());
+    metrics.register(THREADPOOL_ACTIVE, (Gauge<Integer>)() ->  service.getActiveCount());
+    metrics.register(THREADPOOL_SIZE, (Gauge<Integer>)() ->  service.getPoolSize());
+    messageSerdeException = metrics.meter(MESSAGE_SERDE_EXCEPTION);
+    tranmissionException = metrics.meter(MESSAGE_TRANSMISSION_EXCEPTION);
+    tranmissionSuccess = metrics.meter(MESSAGE_TRANSMISSION_SUCCESS);
   }
   
   public void addSharedData(SharedGossipDataMessage message){
@@ -175,29 +189,29 @@
   }
   
   /**
-   * Sends a blocking  message. Throws exception when tranmission fails 
+   * Sends a blocking message.  
    * @param message
    * @param uri
+   * @throws RuntimeException if data can not be serialized or in transmission error
    */
   private void sendInternal(Base message, URI uri){
     byte[] json_bytes;
     try {
       json_bytes = MAPPER.writeValueAsString(message).getBytes();
     } catch (IOException e) {
+      messageSerdeException.mark();
       throw new RuntimeException(e);
     }
-    int packet_length = json_bytes.length;
-    if (packet_length < GossipManager.MAX_PACKET_SIZE) {
-      byte[] buf = UdpUtil.createBuffer(packet_length, json_bytes);
-      try (DatagramSocket socket = new DatagramSocket()) {
-        socket.setSoTimeout(gossipManager.getSettings().getGossipInterval() * 2);
-        InetAddress dest = InetAddress.getByName(uri.getHost());
-        DatagramPacket datagramPacket = new DatagramPacket(buf, buf.length, dest, uri.getPort());
-        socket.send(datagramPacket);
-      } catch (IOException e) {
-        throw new RuntimeException(e);
-      } 
-    }
+    try (DatagramSocket socket = new DatagramSocket()) {
+      socket.setSoTimeout(gossipManager.getSettings().getGossipInterval() * 2);
+      InetAddress dest = InetAddress.getByName(uri.getHost());
+      DatagramPacket datagramPacket = new DatagramPacket(json_bytes, json_bytes.length, dest, uri.getPort());
+      socket.send(datagramPacket);
+      tranmissionSuccess.mark();
+    } catch (IOException e) {
+      tranmissionException.mark();
+      throw new RuntimeException(e);
+    } 
   }
   
   public Response send(Base message, URI uri){
@@ -225,7 +239,7 @@
             return (Response) b;
           }
           try {
-            Thread.sleep(0, 1000);
+            Thread.sleep(0, 555555);
           } catch (InterruptedException e) {
             
           }
@@ -261,19 +275,20 @@
   public void sendOneWay(Base message, URI u){
     byte[] json_bytes;
     try {
-      json_bytes = MAPPER.writeValueAsString(message).getBytes();
+      json_bytes = MAPPER.writeValueAsBytes(message);
     } catch (IOException e) {
+      messageSerdeException.mark();
       throw new RuntimeException(e);
     }
-    int packet_length = json_bytes.length;
-    if (packet_length < GossipManager.MAX_PACKET_SIZE) {
-      byte[] buf = UdpUtil.createBuffer(packet_length, json_bytes);
-      try (DatagramSocket socket = new DatagramSocket()) {
-        socket.setSoTimeout(gossipManager.getSettings().getGossipInterval() * 2);
-        InetAddress dest = InetAddress.getByName(u.getHost());
-        DatagramPacket datagramPacket = new DatagramPacket(buf, buf.length, dest, u.getPort());
-        socket.send(datagramPacket);
-      } catch (IOException ex) { }
+    try (DatagramSocket socket = new DatagramSocket()) {
+      socket.setSoTimeout(gossipManager.getSettings().getGossipInterval() * 2);
+      InetAddress dest = InetAddress.getByName(u.getHost());
+      DatagramPacket datagramPacket = new DatagramPacket(json_bytes, json_bytes.length, dest, u.getPort());
+      socket.send(datagramPacket);
+      tranmissionSuccess.mark();
+    } catch (IOException ex) { 
+      tranmissionException.mark();
+      LOGGER.debug("Send one way failed", ex);
     }
   }
 
diff --git a/src/main/java/org/apache/gossip/manager/GossipCoreConstants.java b/src/main/java/org/apache/gossip/manager/GossipCoreConstants.java
new file mode 100644
index 0000000..6d3765a
--- /dev/null
+++ b/src/main/java/org/apache/gossip/manager/GossipCoreConstants.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.gossip.manager;
+
+public interface GossipCoreConstants {
+  String WORKQUEUE_SIZE = "gossip.core.workqueue.size";
+  String PER_NODE_DATA_SIZE = "gossip.core.pernodedata.size"; 
+  String SHARED_DATA_SIZE = "gossip.core.shareddata.size";
+  String REQUEST_SIZE = "gossip.core.requests.size";
+  String THREADPOOL_ACTIVE = "gossip.core.threadpool.active";
+  String THREADPOOL_SIZE = "gossip.core.threadpool.size";
+  String MESSAGE_SERDE_EXCEPTION = "gossip.core.message_serde_exception";
+  String MESSAGE_TRANSMISSION_EXCEPTION = "gossip.core.message_transmission_exception";
+  String MESSAGE_TRANSMISSION_SUCCESS = "gossip.core.message_transmission_success";
+}
diff --git a/src/main/java/org/apache/gossip/manager/GossipManager.java b/src/main/java/org/apache/gossip/manager/GossipManager.java
index fb7ec93..cf67c9c 100644
--- a/src/main/java/org/apache/gossip/manager/GossipManager.java
+++ b/src/main/java/org/apache/gossip/manager/GossipManager.java
@@ -49,8 +49,6 @@
 
   public static final Logger LOGGER = Logger.getLogger(GossipManager.class);
 
-  public static final int MAX_PACKET_SIZE = 102400;
-
   private final ConcurrentSkipListMap<LocalGossipMember, GossipState> members;
 
   private final LocalGossipMember me;
@@ -82,7 +80,7 @@
           List<GossipMember> gossipMembers, GossipListener listener, MetricRegistry registry) {
     
     this.settings = settings;
-    gossipCore = new GossipCore(this);
+    gossipCore = new GossipCore(this, registry);
     clock = new SystemClock();
     dataReaper = new DataReaper(gossipCore, clock);
     me = new LocalGossipMember(cluster, uri, id, clock.nanoTime(),
@@ -256,4 +254,5 @@
     return dataReaper;
   }
             
+  
 }
diff --git a/src/main/java/org/apache/gossip/manager/PassiveGossipThread.java b/src/main/java/org/apache/gossip/manager/PassiveGossipThread.java
index b54a963..ebda513 100644
--- a/src/main/java/org/apache/gossip/manager/PassiveGossipThread.java
+++ b/src/main/java/org/apache/gossip/manager/PassiveGossipThread.java
@@ -78,23 +78,13 @@
         byte[] buf = new byte[server.getReceiveBufferSize()];
         DatagramPacket p = new DatagramPacket(buf, buf.length);
         server.receive(p);
-        int packet_length = UdpUtil.readPacketLengthFromBuffer(buf);
-        if (packet_length <= GossipManager.MAX_PACKET_SIZE) {
-          byte[] json_bytes = new byte[packet_length];
-          for (int i = 0; i < packet_length; i++) {
-            json_bytes[i] = buf[i + 4];
-          }
-          debug(packet_length, json_bytes);
-          try {
-            Base activeGossipMessage = MAPPER.readValue(json_bytes, Base.class);
-            gossipCore.receive(activeGossipMessage);
-          } catch (RuntimeException ex) {//TODO trap json exception
-            LOGGER.error("Unable to process message", ex);
-          }
-        } else {
-          LOGGER.error("The received message is not of the expected size, it has been dropped.");
+        debug(p.getData());
+        try {
+          Base activeGossipMessage = MAPPER.readValue(p.getData(), Base.class);
+          gossipCore.receive(activeGossipMessage);
+        } catch (RuntimeException ex) {//TODO trap json exception
+          LOGGER.error("Unable to process message", ex);
         }
-
       } catch (IOException e) {
         LOGGER.error(e);
         keepRunning.set(false);
@@ -103,11 +93,10 @@
     shutdown();
   }
 
-  private void debug(int packetLength, byte[] jsonBytes) {
+  private void debug(byte[] jsonBytes) {
     if (LOGGER.isDebugEnabled()){
       String receivedMessage = new String(jsonBytes);
-      LOGGER.debug("Received message (" + packetLength + " bytes): "
-            + receivedMessage);
+      LOGGER.debug("Received message ( bytes): " + receivedMessage);
     }
   }
 
diff --git a/src/main/java/org/apache/gossip/manager/UdpUtil.java b/src/main/java/org/apache/gossip/manager/UdpUtil.java
deleted file mode 100644
index c61769f..0000000
--- a/src/main/java/org/apache/gossip/manager/UdpUtil.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you 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 org.apache.gossip.manager;
-
-import java.nio.ByteBuffer;
-
-public class UdpUtil {
-
-  public static int readPacketLengthFromBuffer(byte [] buffer){
-    int packetLength = 0;
-    for (int i = 0; i < 4; i++) {
-      int shift = (4 - 1 - i) * 8;
-      packetLength += (buffer[i] & 0x000000FF) << shift;
-    }
-    return packetLength;
-  }
-  
-  public static byte[] createBuffer(int packetLength, byte[] jsonBytes) {
-    byte[] lengthBytes = new byte[4];
-    lengthBytes[0] = (byte) (packetLength >> 24);
-    lengthBytes[1] = (byte) ((packetLength << 8) >> 24);
-    lengthBytes[2] = (byte) ((packetLength << 16) >> 24);
-    lengthBytes[3] = (byte) ((packetLength << 24) >> 24);
-    ByteBuffer byteBuffer = ByteBuffer.allocate(4 + jsonBytes.length);
-    byteBuffer.put(lengthBytes);
-    byteBuffer.put(jsonBytes);
-    byte[] buf = byteBuffer.array();
-    return buf;
-  }
-}
diff --git a/src/test/java/org/apache/gossip/manager/DataReaperTest.java b/src/test/java/org/apache/gossip/manager/DataReaperTest.java
index 5388bb3..b4ac45d 100644
--- a/src/test/java/org/apache/gossip/manager/DataReaperTest.java
+++ b/src/test/java/org/apache/gossip/manager/DataReaperTest.java
@@ -32,27 +32,37 @@
 public class DataReaperTest {
 
   private final MetricRegistry registry = new MetricRegistry();
-
+  String myId = "4";
+  String key = "key";
+  String value = "a";
+  
   @Test
   public void testReaperOneShot() {
-    String myId = "4";
-    String key = "key";
-    String value = "a";
     GossipSettings settings = new GossipSettings();
     GossipManager gm = RandomGossipManager.newBuilder().cluster("abc").settings(settings)
             .withId(myId).uri(URI.create("udp://localhost:6000")).registry(registry).build();
     gm.init();
     gm.gossipPerNodeData(perNodeDatum(key, value));
     gm.gossipSharedData(sharedDatum(key, value));
-    Assert.assertEquals(value, gm.findPerNodeGossipData(myId, key).getPayload());
-    Assert.assertEquals(value, gm.findSharedGossipData(key).getPayload());
+    assertDataIsAtCorrectValue(gm);
     gm.getDataReaper().runPerNodeOnce();
     gm.getDataReaper().runSharedOnce();
-    TUnit.assertThat(() -> gm.findPerNodeGossipData(myId, key)).equals(null);
-    TUnit.assertThat(() -> gm.findSharedGossipData(key)).equals(null);
+    assertDataIsRemoved(gm);
     gm.shutdown();
   }
 
+  private void assertDataIsAtCorrectValue(GossipManager gm){
+    Assert.assertEquals(value, gm.findPerNodeGossipData(myId, key).getPayload());
+    Assert.assertEquals(1, registry.getGauges().get(GossipCoreConstants.PER_NODE_DATA_SIZE).getValue());
+    Assert.assertEquals(value, gm.findSharedGossipData(key).getPayload());
+    Assert.assertEquals(1, registry.getGauges().get(GossipCoreConstants.SHARED_DATA_SIZE).getValue());
+  }
+  
+  private void assertDataIsRemoved(GossipManager gm){
+    TUnit.assertThat(() -> gm.findPerNodeGossipData(myId, key)).equals(null);
+    TUnit.assertThat(() -> gm.findSharedGossipData(key)).equals(null);
+  }
+  
   private GossipDataMessage perNodeDatum(String key, String value) {
     GossipDataMessage m = new GossipDataMessage();
     m.setExpireAt(System.currentTimeMillis() + 5L);