@Nullable
    @Override
    public Message decode(@Nonnull RawMessage rawMessage) {
        final byte[] payload = rawMessage.getPayload();
        final JsonNode event;
        try {
            event = objectMapper.readTree(payload);
            if (event == null || event.isMissingNode()) {
                throw new IOException("null result");
            }
        } catch (IOException e) {
            LOG.error("Couldn't decode raw message {}", rawMessage);
            return null;
        }

        return parseEvent(event);
    }
@Test
    public void decodeMessagesHandlesFilebeatMessagesWithoutPrefix() throws Exception {
        configuration = new Configuration(Collections.singletonMap("no_beats_prefix", true));
        codec = new Beats2Codec(configuration, objectMapper, messageFactory);

        final Message message = codec.decode(messageFromJson("filebeat.json"));
        assertThat(message).isNotNull();
        assertThat(message.getMessage()).isEqualTo("TEST");
        assertThat(message.getTimestamp()).isEqualTo(new DateTime(2016, 4, 1, 0, 0, DateTimeZone.UTC));
        assertThat(message.getField("beats_type")).isEqualTo("filebeat");
        assertThat(message.getField("source")).isEqualTo("/tmp/test.log");
        assertThat(message.getField("input_type")).isEqualTo("log");
        assertThat(message.getField("count")).isEqualTo(1);
        assertThat(message.getField("offset")).isEqualTo(0);
        assertThat(message.getField(Message.FIELD_GL2_SOURCE_COLLECTOR)).isEqualTo("1234-5678-1234-5678");
        assertThat(message.getField("filebeat_" + Message.FIELD_GL2_SOURCE_COLLECTOR)).isNull();
        @SuppressWarnings("unchecked") final List<String> tags = (List<String>) message.getField("tags");
        assertThat(tags).containsOnly("foobar", "test");
    }