Awesome

Introduction

This project provides a Serializer, Deserializer and a Serde for Kafka Streams using Jackson for JSON processing. All of the available settings for Jackson are configurable.

What if I don't know what settings to use.

If you are already defining a ObjectMapper that works for you and you want to use it's settings. Use this method to dump the non default settings. This will compare all of the default settings against the supplied ObjectMapper and return the configuration values that the equivalent.

ObjectMapper objectMapper = new ObjectMapper();
objectMapper.configure(SerializationFeature.INDENT_OUTPUT, true)
Map<String, String> actual = JacksonSerializer.nonDefaultSettings(objectMapper);

Kafka Streams Serde

The Serde can be configured by passing in the class to deserialize to. This will return a Serde with configured to the object type. When configure is called it will ignore any output.class supplied since it has already been configured.

Serde<TestPojo> serde = JacksonSerde.of(TestPojo.class);

Serializer

The Serializer does not need to be specified with a type. It will serialize any object that is passed to it.

Map<String, String> settings = ImmutableMap.of(
    ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, JacksonSerializer.class.getName(),
    ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JacksonSerializer.class.getName()
);
Producer<String, TestPojo> producer = new KafkaProducer<>(settings);
// Do your thing

Configuration

Deserializer

Generically processing Json

By default if you do not configure a class to deserialize to the serializer will deserialize to a Jackson JsonNode. Assuming that your data is Json objects you can use ObjectNode.

Map<String, String> settings = ImmutableMap.of(
    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, JacksonDeserializer.class.getName(),
    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JacksonDeserializer.class.getName()
);
Consumer<JsonNode, JsonNode> consumer = new KafkaConsumer<JsonNode, JsonNode>(settings);
// Do your thing

Strongly typed Json

You can tell the deserializer to deserialize to a Pojo.

Map<String, Object> settings = ImmutableMap.of(
    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, JacksonDeserializer.class.getName(),
    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JacksonDeserializer.class.getName(),
    "value.deserializer.output.class", TestPojo.class.getName()
);
Consumer<JsonNode, TestPojo> consumer = new KafkaConsumer<>(settings);

Configuration