@Nullable
    @Override
    public Message decode(@Nonnull RawMessage rawMessage) {
        String s = new String(rawMessage.getPayload(), StandardCharsets.UTF_8);
        LOG.trace("Received raw message: {}", s);

        String timezoneID = configuration.getString(CK_TIMEZONE);
        // previously existing PA inputs after updating will not have a Time Zone configured, default to UTC
        DateTimeZone timezone = timezoneID != null ? DateTimeZone.forID(timezoneID) : DateTimeZone.UTC;
        LOG.trace("Configured time zone: {}", timezone);
        PaloAltoMessageBase p = parser.parse(s, timezone);

        // Return when error occurs parsing syslog header.
        if (p == null) {
            return null;
        }

        Message message = messageFactory.createMessage(p.payload(), p.source(), p.timestamp());

        switch (p.panType()) {
            case "THREAT":
                final PaloAltoTypeParser parserThreat = new PaloAltoTypeParser(templates.getThreatMessageTemplate());
                message.addFields(parserThreat.parseFields(p.fields(), timezone));
                break;
            case "SYSTEM":
                final PaloAltoTypeParser parserSystem = new PaloAltoTypeParser(templates.getSystemMessageTemplate());
                message.addFields(parserSystem.parseFields(p.fields(), timezone));
                break;
            case "TRAFFIC":
                final PaloAltoTypeParser parserTraffic = new PaloAltoTypeParser(templates.getTrafficMessageTemplate());
                message.addFields(parserTraffic.parseFields(p.fields(), timezone));
                break;
            default:
                LOG.error("Unsupported PAN type [{}]. Not adding any parsed fields.", p.panType());
        }

        LOG.trace("Successfully processed [{}] message with [{}] fields.", p.panType(), message.getFieldCount());

        return message;
    }
@Test
    public void testMessageWithLineBreak() {

        // Verify that a messages with a line break at the end does not break parsing.
        PaloAltoCodec codec = new PaloAltoCodec(Configuration.EMPTY_CONFIGURATION, messageFactory);
        Message message = codec.decode(new RawMessage(PANORAMA_WITH_LINE_BREAK.getBytes(StandardCharsets.UTF_8)));
        assertEquals("SYSTEM", message.getField("type"));

        codec = new PaloAltoCodec(Configuration.EMPTY_CONFIGURATION, messageFactory);
        message = codec.decode(new RawMessage(SYSLOG_WITH_LINE_BREAK.getBytes(StandardCharsets.UTF_8)));
        assertEquals("THREAT", message.getField("type"));
    }