class FlowFileAttributesSerializer implements Deserializer<Map<String, String>>, Serializer<Map<String, String>> {

    @Override
    public Map<String, String> deserialize(final byte[] input) throws DeserializationException, IOException {
        if (input == null || input.length == 0) {
            return null;
        }
        Map<String, String> attributes = new HashMap<>();

        String attributesText = new String(input, StandardCharsets.UTF_8);
        String[] entries = attributesText.split(Pattern.quote(ATTRIBUTE_SEPARATOR));
        for(String entry : entries) {
            int equalsIndex = entry.indexOf('=');
            String key = entry.substring(0, equalsIndex);
            String value = entry.substring(equalsIndex + 1);
            attributes.put(key, value);
        }

        return attributes;
    }

}