public Set<Flow> parseDataSet(ImmutableList<InformationElement> informationElements, Map<Integer, TemplateRecord> templateMap, ByteBuf setContent) {
        ImmutableSet.Builder<Flow> flowBuilder = ImmutableSet.builder();

        // Padding: data records have no header and no fixed length, but the end of a data set (which are non-delimited series of data records), there _may_ be padding
        // (see https://datatracker.ietf.org/doc/html/rfc7011#section-3.3.1)
        // Furthermore it states: "The padding length MUST be shorter than any allowable record in this Set."
        // We could calculate the minimum record length from informationElements, but we would need to take into consideration that it might be made up of variable length elements.
        // Thus, it is much simpler to catch the exception and return the flows we have collected so far.

        int flowNum = 1;
        readFlows:
        while (setContent.isReadable()) {
            try {
                if (LOG.isTraceEnabled()) {
                    LOG.trace("Parsing flow {}", flowNum);
                }
                flowNum++;
                final ImmutableMap.Builder<String, Object> fields = ImmutableMap.builder();
                for (InformationElement informationElement : informationElements) {
                    final int firstByte = setContent.readerIndex();
                    InformationElementDefinition desc = infoElemDefs.getDefinition(informationElement.id(), informationElement.enterpriseNumber());
                    switch (desc.dataType()) {
                        // these are special because they can use reduced-size encoding (RFC 7011 Sec 6.2)
                        case UNSIGNED8:
                        case UNSIGNED16:
                        case UNSIGNED32:
                        case UNSIGNED64:
                            long unsignedValue;
                            switch (informationElement.length()) {
                                case 1:
                                    unsignedValue = setContent.readUnsignedByte();
                                    break;
                                case 2:
                                    unsignedValue = setContent.readUnsignedShort();
                                    break;
                                case 3:
                                    unsignedValue = setContent.readUnsignedMedium();
                                    break;
                                case 4:
                                    unsignedValue = setContent.readUnsignedInt();
                                    break;
                                case 5:
                                case 6:
                                case 7:
                                case 8:
                                    byte[] bytesBigEndian = {0, 0, 0, 0, 0, 0, 0, 0};
                                    int firstIndex = 8 - informationElement.length();
                                    setContent.readBytes(bytesBigEndian, firstIndex, informationElement.length());
                                    unsignedValue = Longs.fromByteArray(bytesBigEndian);
                                    break;
                                default:
                                    throw new IpfixException("Unexpected length for unsigned integer");
                            }
                            fields.put(desc.fieldName(), unsignedValue);
                            break;
                        case SIGNED8:
                        case SIGNED16:
                        case SIGNED32:
                        case SIGNED64:
                            long signedValue;
                            switch (informationElement.length()) {
                                case 1:
                                    signedValue = setContent.readByte();
                                    break;
                                case 2:
                                    signedValue = setContent.readShort();
                                    break;
                                case 3:
                                    signedValue = setContent.readMedium();
                                    break;
                                case 4:
                                    signedValue = setContent.readUnsignedInt();
                                    break;
                                case 5:
                                case 6:
                                case 7:
                                case 8:
                                    byte[] bytesBigEndian = {0, 0, 0, 0, 0, 0, 0, 0};
                                    int firstIndex = 8 - informationElement.length() - 1;
                                    setContent.readBytes(bytesBigEndian, firstIndex, informationElement.length());
                                    signedValue = Longs.fromByteArray(bytesBigEndian);
                                    break;
                                default:
                                    throw new IpfixException("Unexpected length for unsigned integer");
                            }
                            fields.put(desc.fieldName(), signedValue);
                            break;
                        case FLOAT32:
                        case FLOAT64:
                            double floatValue;
                            switch (informationElement.length()) {
                                case 4:
                                    floatValue = setContent.readFloat();
                                    break;
                                case 8:
                                    floatValue = setContent.readDouble();
                                    break;
                                default:
                                    throw new IpfixException("Unexpected length for float value: " + informationElement.length());
                            }
                            fields.put(desc.fieldName(), floatValue);
                            break;
                        // the remaining types aren't subject to reduced-size encoding
                        case MACADDRESS:
                            byte[] macBytes = new byte[6];
                            setContent.readBytes(macBytes);
                            fields.put(desc.fieldName(),
                                    String.format(Locale.ROOT, "%02x:%02x:%02x:%02x:%02x:%02x",
                                            macBytes[0], macBytes[1], macBytes[2], macBytes[3], macBytes[4], macBytes[5]));

                            break;
                        case IPV4ADDRESS:
                            byte[] ipv4Bytes = new byte[4];
                            setContent.readBytes(ipv4Bytes);
                            try {
                                fields.put(desc.fieldName(), InetAddress.getByAddress(ipv4Bytes).getHostAddress());
                            } catch (UnknownHostException e) {
                                throw new IpfixException("Unable to parse IPV4 address", e);
                            }
                            break;
                        case IPV6ADDRESS:
                            byte[] ipv6Bytes = new byte[16];
                            setContent.readBytes(ipv6Bytes);
                            try {
                                fields.put(desc.fieldName(), InetAddress.getByAddress(ipv6Bytes).getHostAddress());
                            } catch (UnknownHostException e) {
                                throw new IpfixException("Unable to parse IPV6 address", e);
                            }
                            break;
                        case BOOLEAN:
                            final byte booleanByte = setContent.readByte();
                            switch (booleanByte) {
                                case 1:
                                    fields.put(desc.fieldName(), true);
                                    break;
                                case 2:
                                    fields.put(desc.fieldName(), false);
                                    break;
                                default:
                                    throw new IpfixException("Invalid value for boolean: " + booleanByte);
                            }
                            break;
                        case STRING:
                            final CharSequence charSequence;
                            if (informationElement.length() == 65535) {
                                // variable length element, parse accordingly
                                int length = getVarLength(setContent);
                                charSequence = setContent.readCharSequence(length, StandardCharsets.UTF_8);
                            } else {
                                // fixed length element, just read the string from the buffer
                                charSequence = setContent.readCharSequence(informationElement.length(), StandardCharsets.UTF_8);
                            }

                            fields.put(desc.fieldName(), String.valueOf(charSequence).replace("\0", ""));
                            break;
                        case OCTETARRAY:
                            final byte[] octetArray;
                            if (informationElement.length() == 65535) {
                                int length = getVarLength(setContent);
                                octetArray = new byte[length];
                            } else {
                                octetArray = new byte[informationElement.length()];
                            }
                            setContent.readBytes(octetArray);
                            fields.put(desc.fieldName(), Hex.encodeHexString(octetArray));
                            break;
                        case DATETIMESECONDS:
                            final long dateTimeSeconds = setContent.readUnsignedInt();
                            fields.put(desc.fieldName(), ZonedDateTime.ofInstant(Instant.ofEpochSecond(dateTimeSeconds), ZoneOffset.UTC));
                            break;
                        case DATETIMEMILLISECONDS:
                            final long dateTimeMills = setContent.readLong();
                            fields.put(desc.fieldName(), ZonedDateTime.ofInstant(Instant.ofEpochMilli(dateTimeMills), ZoneOffset.UTC));
                            break;
                        case DATETIMEMICROSECONDS:
                        case DATETIMENANOSECONDS:
                            final long seconds = setContent.readUnsignedInt();
                            long fraction = setContent.readUnsignedInt();
                            if (desc.dataType() == InformationElementDefinition.DataType.DATETIMEMICROSECONDS) {
                                // bottom 11 bits must be cleared for micros to ensure the precision is correct (RFC 7011 Sec 6.1.9)
                                fraction = fraction & ~0x7FF;
                            }
                            fields.put(desc.fieldName(), ZonedDateTime.ofInstant(Instant.ofEpochSecond(seconds, fraction), ZoneOffset.UTC));
                            break;
                        case BASICLIST: {
                            // TODO add to field somehow
                            int length = informationElement.length() == 65535 ? getVarLength(setContent) : setContent.readUnsignedByte();
                            ByteBuf listBuffer = setContent.readSlice(length);
                            final short semantic = listBuffer.readUnsignedByte();
                            final InformationElement element = parseInformationElement(listBuffer);
                            InformationElementDefinition def = infoElemDefs.getDefinition(element.id(), element.enterpriseNumber());
                            if (def == null) {
                                LOG.error("Unable to find information element definition in basicList: id {} PEN {}, this is a bug, cannot parse packet.", element.id(), element.enterpriseNumber());
                                break;
                            } else {
                                LOG.warn("Skipping basicList data ({} bytes)", informationElement.length());
                                while (listBuffer.isReadable()) {
                                    // simply discard the bytes for now
                                    listBuffer.skipBytes(element.length());
                                }
                            }
                            break;
                        }
                        case SUBTEMPLATELIST: {
                            // there are three possibilities here (compare https://tools.ietf.org/html/rfc6313#section-4.5.2):
                            //  1. the data set's template has an explicit length
                            //  2. the length is < 255 encoded as 1 byte, in variable length format (not recommended)
                            //  3. the length is encoded as 3 bytes, in variable length format (recommended per RFC 6313)
                        /* encoding format in this case is according to Figure 5:
                            0                   1                   2                   3
                            0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
                           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                           |   Semantic    |         Template ID           |     ...       |
                           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                           |                subTemplateList Content    ...                 |
                           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                           |                              ...                              |
                           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
                          Semantic is one of:
                            * 0xFF - undefined
                            * 0x00 - noneOf
                            * 0x01 - exactlyOneOf
                            * 0x02 - oneOrMoreOf
                            * 0x03 - allOf
                            * 0x04 - ordered
                         */
                            int length = informationElement.length() == 65535 ? getVarLength(setContent) : setContent.readUnsignedByte();
                            // adjust length for semantic + templateId
                            length -= 3;
                            LOG.debug("Remaining data buffer:\n{}", ByteBufUtil.prettyHexDump(setContent));
                            // TODO add to field somehow
                            final short semantic = setContent.readUnsignedByte();
                            final int templateId = setContent.readUnsignedShort();
                            final TemplateRecord templateRecord = templateMap.get(templateId);
                            if (templateRecord == null) {
                                LOG.error("Unable to parse subtemplateList, because we don't have the template for it: {}, skipping data ({} bytes)", templateId, length);
                                setContent.skipBytes(length);
                                break;
                            }
                            final ByteBuf listContent = setContent.readSlice(length);
                            // if this is not readable, it's an empty list
                            final ImmutableList.Builder<Flow> flowsBuilder = ImmutableList.builder();
                            if (listContent.isReadable()) {
                                flowsBuilder.addAll(parseDataSet(templateRecord.informationElements(), templateMap, listContent));
                            }
                            final ImmutableList<Flow> flows = flowsBuilder.build();
                            // flatten arrays and fields into the field name until we have support for nested objects
                            for (int i = 0; i < flows.size(); i++) {
                                final String fieldPrefix = desc.fieldName() + "_" + i + "_";
                                flows.get(i).fields().forEach((field, value) -> {
                                    fields.put(fieldPrefix + field, value);
                                });
                            }
                            break;
                        }
                        case SUBTEMPLATEMULTILIST: {
                            int length = informationElement.length() == 65535 ? getVarLength(setContent) : setContent.readUnsignedByte();
                            setContent.skipBytes(length);
                            LOG.warn("subtemplateMultilist support is not implemented, skipping data ({} bytes)", length);
                            break;
                        }
                    }
                    if (LOG.isTraceEnabled()) {
                        LOG.trace("Field {}:\n{}", informationElement.id(), ByteBufUtil.prettyHexDump(setContent, firstByte, setContent.readerIndex() - firstByte));
                    }
                }
                flowBuilder.add(Flow.create(fields.build()));
            } catch (IndexOutOfBoundsException ioobe) {
                // this can happen if we read past the valid flows into the optional padding following a record set.
                // to differentiate a bug and the legitimate trailing 0x00 bytes padding case, we actually consume the rest of the bytes and check that they are all 0.
                if (setContent.forEachByte(value -> value == 0x00) != -1) {
                    // if we still have bytes left, it means we had non-zero bytes, thus not only padding. this is bad and probably a bug of some sort.
                    LOG.error("Verify data record padding failed, trailing bytes were not all 0x00");
                    throw ioobe;
                }
                // label for clarity
                //noinspection UnnecessaryLabelOnBreakStatement
                break readFlows;
            }
        }
        return flowBuilder.build();
    }
@Test
    public void parseDataSet() throws IOException {

        final ByteBuf packet = Utils.readPacket("templates-data.ipfix");
        InformationElementDefinitions infoElementDefs = new InformationElementDefinitions(Resources.getResource("ipfix-iana-elements.json"),
                                                                                          Resources.getResource("ixia-ied.json"));

        final IpfixMessage ipfixMessage = new IpfixParser(infoElementDefs).parseMessage(packet);
        assertThat(ipfixMessage).isNotNull();

        final ImmutableList<TemplateRecord> templateRecords = ipfixMessage.templateRecords();
        assertThat(templateRecords).hasSize(2);

        final TemplateRecord firstTemplate = templateRecords.get(0);
        assertThat(firstTemplate.templateId()).isEqualTo(256);
        assertThat(firstTemplate.informationElements()).hasSize(51)
                                                       .containsExactly(
                                                               InformationElement.create(1, 8, 0),
                                                               InformationElement.create(2, 8, 0),
                                                               InformationElement.create(4, 1, 0),
                                                               InformationElement.create(6, 1, 0),
                                                               InformationElement.create(7, 2, 0),
                                                               InformationElement.create(8, 4, 0),
                                                               InformationElement.create(10, 4, 0),
                                                               InformationElement.create(11, 2, 0),
                                                               InformationElement.create(12, 4, 0),
                                                               InformationElement.create(14, 4, 0),
                                                               InformationElement.create(16, 4, 0),
                                                               InformationElement.create(17, 4, 0),
                                                               InformationElement.create(32, 2, 0),
                                                               InformationElement.create(136, 1, 0),
                                                               InformationElement.create(152, 8, 0),
                                                               InformationElement.create(153, 8, 0),
                                                               InformationElement.create(110, 4, 3054),
                                                               InformationElement.create(111, 65535, 3054),
                                                               InformationElement.create(120, 4, 3054),
                                                               InformationElement.create(121, 65535, 3054),
                                                               InformationElement.create(122, 4, 3054),
                                                               InformationElement.create(123, 65535, 3054),
                                                               InformationElement.create(125, 65535, 3054),
                                                               InformationElement.create(126, 4, 3054),
                                                               InformationElement.create(127, 4, 3054),
                                                               InformationElement.create(140, 4, 3054),
                                                               InformationElement.create(141, 65535, 3054),
                                                               InformationElement.create(142, 4, 3054),
                                                               InformationElement.create(143, 65535, 3054),
                                                               InformationElement.create(145, 65535, 3054),
                                                               InformationElement.create(146, 4, 3054),
                                                               InformationElement.create(147, 4, 3054),
                                                               InformationElement.create(160, 1, 3054),
                                                               InformationElement.create(161, 65535, 3054),
                                                               InformationElement.create(162, 1, 3054),
                                                               InformationElement.create(163, 65535, 3054),
                                                               InformationElement.create(178, 65535, 3054),
                                                               InformationElement.create(179, 65535, 3054),
                                                               InformationElement.create(180, 2, 3054),
                                                               InformationElement.create(182, 65535, 3054),
                                                               InformationElement.create(183, 65535, 3054),
                                                               InformationElement.create(184, 65535, 3054),
                                                               InformationElement.create(185, 65535, 3054),
                                                               InformationElement.create(186, 65535, 3054),
                                                               InformationElement.create(187, 65535, 3054),
                                                               InformationElement.create(188, 4, 3054),
                                                               InformationElement.create(189, 65535, 3054),
                                                               InformationElement.create(190, 65535, 3054),
                                                               InformationElement.create(191, 65535, 3054),
                                                               InformationElement.create(192, 65535, 3054),
                                                               InformationElement.create(193, 4, 3054)
                                                       );

        final TemplateRecord secondTemplate = templateRecords.get(1);
        assertThat(secondTemplate.templateId()).isEqualTo(257);
        assertThat(secondTemplate.informationElements())
                .hasSize(50)
                .containsOnly(
                        InformationElement.create(1, 8, 0),
                        InformationElement.create(2, 8, 0),
                        InformationElement.create(4, 1, 0),
                        InformationElement.create(6, 1, 0),
                        InformationElement.create(7, 2, 0),
                        InformationElement.create(10, 4, 0),
                        InformationElement.create(11, 2, 0),
                        InformationElement.create(14, 4, 0),
                        InformationElement.create(16, 4, 0),
                        InformationElement.create(17, 4, 0),
                        InformationElement.create(27, 16, 0),
                        InformationElement.create(28, 16, 0),
                        InformationElement.create(136, 1, 0),
                        InformationElement.create(139, 2, 0),
                        InformationElement.create(152, 8, 0),
                        InformationElement.create(153, 8, 0),
                        InformationElement.create(110, 4, 3054),
                        InformationElement.create(111, 65535, 3054),
                        InformationElement.create(120, 4, 3054),
                        InformationElement.create(121, 65535, 3054),
                        InformationElement.create(122, 4, 3054),
                        InformationElement.create(123, 65535, 3054),
                        InformationElement.create(125, 65535, 3054),
                        InformationElement.create(126, 4, 3054),
                        InformationElement.create(127, 4, 3054),
                        InformationElement.create(140, 4, 3054),
                        InformationElement.create(141, 65535, 3054),
                        InformationElement.create(142, 4, 3054),
                        InformationElement.create(143, 65535, 3054),
                        InformationElement.create(145, 65535, 3054),
                        InformationElement.create(146, 4, 3054),
                        InformationElement.create(147, 4, 3054),
                        InformationElement.create(160, 1, 3054),
                        InformationElement.create(161, 65535, 3054),
                        InformationElement.create(162, 1, 3054),
                        InformationElement.create(163, 65535, 3054),
                        InformationElement.create(178, 65535, 3054),
                        InformationElement.create(179, 65535, 3054),
                        InformationElement.create(180, 2, 3054),
                        InformationElement.create(182, 65535, 3054),
                        InformationElement.create(183, 65535, 3054),
                        InformationElement.create(184, 65535, 3054),
                        InformationElement.create(185, 65535, 3054),
                        InformationElement.create(186, 65535, 3054),
                        InformationElement.create(187, 65535, 3054),
                        InformationElement.create(188, 4, 3054),
                        InformationElement.create(189, 65535, 3054),
                        InformationElement.create(190, 65535, 3054),
                        InformationElement.create(191, 65535, 3054),
                        InformationElement.create(192, 65535, 3054)
                );

        final ImmutableList<OptionsTemplateRecord> optionsTemplateRecords = ipfixMessage.optionsTemplateRecords();
        assertThat(optionsTemplateRecords).describedAs("options template records").hasSize(1);
        assertThat(optionsTemplateRecords.get(0).scopeFields())
                .describedAs("first record's scope fields")
                .hasSize(1);
        assertThat(ipfixMessage.flows()).describedAs("Flows")
                                        .hasSize(1);
        final Flow flow = ipfixMessage.flows().get(0);
        assertThat(flow.fields()).containsExactly(
                immutableEntry("octetDeltaCount", 103L),
                immutableEntry("packetDeltaCount", 1L),
                immutableEntry("protocolIdentifier", 17L),
                immutableEntry("tcpControlBits", 0L),
                immutableEntry("sourceTransportPort", 53L),
                immutableEntry("sourceIPv4Address", "36.83.97.149"),
                immutableEntry("ingressInterface", 1L),
                immutableEntry("destinationTransportPort", 30297L),
                immutableEntry("destinationIPv4Address", "36.83.97.7"),
                immutableEntry("egressInterface", 1L),
                immutableEntry("bgpSourceAsNumber", 17974L),
                immutableEntry("bgpDestinationAsNumber", 17974L),
                immutableEntry("icmpTypeCodeIPv4", 0L),
                immutableEntry("flowEndReason", 1L),
                immutableEntry("flowStartMilliseconds", ZonedDateTime.of(2018, 9, 13, 21, 39, 13, 249_000_000, ZoneOffset.UTC)),
                immutableEntry("flowEndMilliseconds", ZonedDateTime.of(2018, 9, 13, 21, 39, 13, 249_000_000, ZoneOffset.UTC)),
                immutableEntry("l7ApplicationId", 1L),
                immutableEntry("l7ApplicationName", "domain"),
                immutableEntry("sourceIpCountryCode", "ID"),
                immutableEntry("sourceIpCountryName", "Indonesia"),
                immutableEntry("sourceIpRegionCode", "SN"),
                immutableEntry("sourceIpRegionName", "South Sulawesi"),
                immutableEntry("sourceIpCityName", "Ballaparang"),
                immutableEntry("sourceIpLatitude", -5.146399974822998),
                immutableEntry("sourceIpLongitude", 119.44129943847656),
                immutableEntry("destinationIpCountryCode", "ID"),
                immutableEntry("destinationIpCountryName", "Indonesia"),
                immutableEntry("destinationIpRegionCode", "SN"),
                immutableEntry("destinationIpRegionName", "South Sulawesi"),
                immutableEntry("destinationIpCityName", "Ballaparang"),
                immutableEntry("destinationIpLatitude", -5.146399974822998),
                immutableEntry("destinationIpLongitude", 119.44129943847656),
                immutableEntry("osDeviceId", 0L),
                immutableEntry("osDeviceName", "unknown"),
                immutableEntry("browserId", 0L),
                immutableEntry("browserName", "-"),
                immutableEntry("sslConnectionEncryptionType", "Cleartext"),
                immutableEntry("sslEncryptionCipherName", "none"),
                immutableEntry("sslEncryptionKeyLength", 0L),
                immutableEntry("userAgent", ""),
                immutableEntry("hostName", ""),
                immutableEntry("uri", ""),
                immutableEntry("dnsText", ""),
                immutableEntry("sourceAsName", "TELKOMNET-AS2-AP PT Telekomunikasi Indonesia, ID"),
                immutableEntry("destinationAsName", "TELKOMNET-AS2-AP PT Telekomunikasi Indonesia, ID"),
                immutableEntry("transactionLatency", 53L),
                immutableEntry("dnsQueryHostName", "server-2453614b.example.int."),
                immutableEntry("dnsResponseHostName", "server-2453614b.example.int."),
                immutableEntry("dnsClasses", "IN"),
                immutableEntry("threatType", ""),
                immutableEntry("threatIpv4", "0.0.0.0")
        );
    }