### Question:
KafkaAreciboClient implements RealtimeClient<Message> { @VisibleForTesting static Event convertByteBufferPayloadToEvent(final byte[] bytes) throws IOException { return mapper.readValue(bytes, MapEvent.class); } KafkaAreciboClient(final RealtimeClientConfig config); @Override void listenToStream(final String topic, final Function<Event, Void> onValidMessage, final Function<Message, Void> onCorruptedMessage); @Override synchronized void stopListening(final String topic); @MonitorableManaged(description = "Number of messages received", monitored = true, monitoringType = {MonitoringType.COUNTER, MonitoringType.RATE}) long getMessagesReceived(); @MonitorableManaged(description = "Number of valid messages received", monitored = true, monitoringType = {MonitoringType.COUNTER, MonitoringType.RATE}) long getValidMessagesReceived(); @MonitorableManaged(description = "Number of messages received that couldn't be parsed into an Event ", monitored = true, monitoringType = {MonitoringType.COUNTER, MonitoringType.RATE}) long getUnParseableMessagesReceived(); @MonitorableManaged(description = "Number of corrupted (invalid crc) messages received", monitored = true, monitoringType = {MonitoringType.COUNTER, MonitoringType.RATE}) long getCorruptedMessagesReceived();  }### Answer:
@Test(groups = "fast") public void testParser() throws Exception { final Event event = KafkaAreciboClient.convertByteBufferPayloadToEvent(payload); Assert.assertEquals(event.getEventType(), "fcc83090-a6d0-4509-8a0a-3a06b99f96f7"); Assert.assertEquals(event.getTimestamp(), 1331148118082L); Assert.assertEquals(event.getSourceUUID(), UUID.fromString("714be299-b735-4663-8efe-74f10dc62a7a")); Assert.assertTrue(event instanceof MapEvent); final MapEvent mapEvent = (MapEvent) event; Assert.assertEquals(Double.valueOf(mapEvent.getMap().get("min_heapUsed").toString()), 1.515698888E9); Assert.assertEquals(Double.valueOf(mapEvent.getMap().get("max_heapUsed").toString()), 1.835511784E9); }