@Nullable
    @Override
    public Message decode(@Nonnull RawMessage rawMessage) {
        final byte[] payload = rawMessage.getPayload();
        final Map<String, Object> event;
        try {
            event = objectMapper.readValue(payload, TypeReferences.MAP_STRING_OBJECT);
        } catch (IOException e) {
            LOG.error("Couldn't decode raw message {}", rawMessage);
            return null;
        }

        return parseEvent(event);
    }
@Test
    public void decodeMessagesHandlesGenericBeatWithKubernetes() throws Exception {
        final Message message = codec.decode(messageFromJson("generic-with-kubernetes.json"));
        assertThat(message).isNotNull();
        assertThat(message.getMessage()).isEqualTo("null");
        assertThat(message.getSource()).isEqualTo("unknown");
        assertThat(message.getTimestamp()).isEqualTo(new DateTime(2016, 4, 1, 0, 0, DateTimeZone.UTC));
        assertThat(message.getField("facility")).isEqualTo("genericbeat");
        assertThat(message.getField("beat_foo")).isEqualTo("bar");
        assertThat(message.getField("beat_kubernetes_pod_name")).isEqualTo("testpod");
        assertThat(message.getField("beat_kubernetes_namespace")).isEqualTo("testns");
        assertThat(message.getField("beat_kubernetes_labels_labelkey")).isEqualTo("labelvalue");
    }