MetricsCountAndSumSerde.java

static class MetricsCountAndSumSerde implements Serde<MetricsCountAndSum> {

        @Override
        public void configure(Map<String, ?> configs, boolean isKey) {
        }

        @Override
        public void close() {
        }

        @Override
        public Serializer<MetricsCountAndSum> serializer() {
            return new Serializer<MetricsCountAndSum>() {
                @Override
                public void configure(Map<String, ?> configs, boolean isKey) {
                }

                @Override
                public byte[] serialize(String topic, MetricsCountAndSum data) {
                    String countAndSum = data.getCount() + ":" + data.getSum();
                    System.out.println("serialized MetricsCountAndSum " + data + " to bytes");
                    return countAndSum.getBytes();
                }

                @Override
                public void close() {
                }
            };
        }

        @Override
        public Deserializer<MetricsCountAndSum> deserializer() {
            return new Deserializer<MetricsCountAndSum>() {
                @Override
                public void configure(Map<String, ?> configs, boolean isKey) {
                }

                @Override
                public MetricsCountAndSum deserialize(String topic, byte[] countAndSum) {
                    String countAndSumStr = new String(countAndSum);
                    Long count = Long.valueOf(countAndSumStr.split(":")[0]);
                    Long sum = Long.valueOf(countAndSumStr.split(":")[1]);

                    MetricsCountAndSum countAndSumObject = new MetricsCountAndSum(count, sum);
                    System.out.println("Deserialized byte[] to MetricsCountAndSum " + countAndSumObject);

                    return countAndSumObject;
                }

                @Override
                public void close() {
                }
            };
        }

    }