@Nullable
    @Override
    public Collection<Message> decodeMessages(@Nonnull RawMessage rawMessage) {
        try {
            final ResolvableInetSocketAddress remoteAddress = rawMessage.getRemoteAddress();
            final InetSocketAddress sender = remoteAddress != null ? remoteAddress.getInetSocketAddress() : null;

            final byte[] payload = rawMessage.getPayload();
            if (payload.length < 3) {
                LOG.debug("NetFlow message (source: {}) doesn't even fit the NetFlow version (size: {} bytes)",
                        sender, payload.length);
                return null;
            }

            final ByteBuf buffer = Unpooled.wrappedBuffer(payload);
            switch (buffer.readByte()) {
                case PASSTHROUGH_MARKER:
                    final NetFlowV5Packet netFlowV5Packet = NetFlowV5Parser.parsePacket(buffer);

                    return netFlowV5Packet.records().stream()
                            .map(record -> netFlowFormatter.toMessage(netFlowV5Packet.header(), record, sender))
                            .collect(Collectors.toList());
                case ORDERED_V9_MARKER:
                    // our "custom" netflow v9 that has all the templates in the same packet
                    return decodeV9(sender, buffer);
                default:
                    final List<RawMessage.SourceNode> sourceNodes = rawMessage.getSourceNodes();
                    final RawMessage.SourceNode sourceNode = sourceNodes.isEmpty() ? null : sourceNodes.get(sourceNodes.size() - 1);
                    final String inputId = sourceNode == null ? "<unknown>" : sourceNode.inputId;
                    LOG.warn("Unsupported NetFlow packet on input {} (source: {})", inputId, sender);
                    return null;
            }
        } catch (FlowException e) {
            LOG.error("Error parsing NetFlow packet <{}> received from <{}>", rawMessage.getId(), rawMessage.getRemoteAddress(), e);
            if (LOG.isDebugEnabled()) {
                LOG.debug("NetFlow packet hexdump:\n{}", ByteBufUtil.prettyHexDump(Unpooled.wrappedBuffer(rawMessage.getPayload())));
            }
            return null;
        } catch (InvalidProtocolBufferException e) {
            LOG.error("Invalid NetFlowV9 entry found, cannot parse the messages", ExceptionUtils.getRootCause(e));
            return null;
        }
    }
@Test
    public void decodeMessagesThrowsEmptyTemplateExceptionWithIncompleteNetFlowV9() throws Exception {
        final byte[] b = Resources.toByteArray(Resources.getResource("netflow-data/netflow-v9-3_incomplete.dat"));
        final InetSocketAddress source = new InetSocketAddress(InetAddress.getLocalHost(), 12345);

        assertThat(codec.decodeMessages(new RawMessage(b, source))).isNull();
    }