Spring Boot Kafka Json Serializer Example, easy in 15 minutes

Today, I introduce a Spring Boot Kafka Json Serializer Example and demo how to send and receive a Java Object as JSON object from Apache Kafka using Spring-Kafka and Spring Boot. I make sure that this tutorial is useful with a real project. Let’ do together.

Contents hide

1 Spring Boot Kafka Json Serializer Example

1.1 Project Setup

1.2 Project Structure

1.3 Maven Dependencies

1.4 Configure Spring Kafka Application with application.yml

1.5 Simple POJO to Serialize/Deserialize

1.6 Using Jackson to Serialize/Deserialize

1.7 Create Kafka Producer and Consumer

1.8 Spring Boot Application

1.9 Running Spring Boot Application

1.10 Download the complete source code

1.11 References

Spring Boot Kafka Json Serializer Example

Project Setup

Spring Kafka: 2.3.4.RELEASE
Spring Boot: 2.2.2.RELEASE
Apache Kafka: kafka_2.13-2.4.0
Maven: 3.5

Project Structure

Spring Boot Kafka Json Serializer Example

Maven Dependencies

At least, we must have these dependencies: spring-boot-starter, spring-kafka, jackson-databind in pom.xml file. Note, you can change spring-kafka version for your own needs.

<properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <spring-kafka.version>2.3.4.RELEASE</spring-kafka.version>
</properties>

<dependencies>        
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>${spring-kafka.version}</version>
        </dependency>

        <!-- json support -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
</dependencies>

<java.version>1.8</java.version>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<spring-kafka.version>2.3.4.RELEASE</spring-kafka.version>

</properties>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter</artifactId>

</dependency>

<groupId>org.springframework.kafka</groupId>

<artifactId>spring-kafka</artifactId>

<version>${spring-kafka.version}</version>

</dependency>

<groupId>com.fasterxml.jackson.core</groupId>

<artifactId>jackson-databind</artifactId>

</dependency>

</dependencies>

Configure Spring Kafka Application with application.yml

Some properties that need to be explicitly set in the application.yml file, e.g host, port, topic name:

spring:
  kafka:
    servers: localhost:9092

app:
  topic:
    staff: staff.t

spring:

kafka:

servers: localhost:9092

app:

topic:

staff: staff.t

Simple POJO to Serialize/Deserialize

We do with simple POJO class named Staff, its object will be serialized/deserialized to java object/json object.

package com.javabycode.kafka;

public class Staff {

    private String name;
    private String country;
    @JsonSerialize(using = LocalDateTimeSerializer.class)
    @JsonDeserialize(using = LocalDateTimeDeserializer.class)
    private LocalDateTime birthday;

    public Staff() {
    }

    public Staff(String name, String country, LocalDateTime birthday) {
        this.name = name;
        this.country = country;
        this.birthday = birthday;
    }
    // getter/setter
}

package com.javabycode.kafka;

public class Staff {

private String name;

private String country;

@JsonSerialize(using = LocalDateTimeSerializer.class)

@JsonDeserialize(using = LocalDateTimeDeserializer.class)

private LocalDateTime birthday;

public Staff() {

}

public Staff(String name, String country, LocalDateTime birthday) {

this.name = name;

this.country = country;

this.birthday = birthday;

}

// getter/setter

}

Using Jackson to Serialize/Deserialize

Here, we create two classes to custom Serialize/Deserialize for the birthday field.

// convert LocalDateTime to string in format MM/dd/yyyy HH:mm:ss
public class LocalDateTimeSerializer extends JsonSerializer<LocalDateTime> {
	    
    private static final Logger LOG = LoggerFactory.getLogger(LocalDateTimeSerializer.class);
	
	@Override
	public void serialize(LocalDateTime arg0, JsonGenerator arg1, SerializerProvider arg2) throws IOException {
		DateTimeFormatter formatter = DateTimeFormatter.ofPattern("MM/dd/yyyy HH:mm:ss");
		String format = formatter.format(arg0);
        LOG.info("Serialized Date='{}'", format);
		arg1.writeString(format);
	}
}

//and 
// convert string in format MM/dd/yyyy HH:mm:ss to LocalDateTime
public class LocalDateTimeDeserializer extends JsonDeserializer<LocalDateTime> {
	
    private static final Logger LOG = LoggerFactory.getLogger(LocalDateTimeDeserializer.class);
	
    @Override
    public LocalDateTime deserialize(JsonParser arg0, DeserializationContext arg1) throws IOException {
    	DateTimeFormatter formatter = DateTimeFormatter.ofPattern("MM/dd/yyyy HH:mm:ss");
    	LOG.info("Deserialized Date='{}'", arg0.getText());
        return LocalDateTime.parse(arg0.getText(),formatter);
    }
}

// convert LocalDateTime to string in format MM/dd/yyyy HH:mm:ss

public class LocalDateTimeSerializer extends JsonSerializer<LocalDateTime> {

private static final Logger LOG = LoggerFactory.getLogger(LocalDateTimeSerializer.class);

@Override

public void serialize(LocalDateTime arg0, JsonGenerator arg1, SerializerProvider arg2) throws IOException {

DateTimeFormatter formatter = DateTimeFormatter.ofPattern("MM/dd/yyyy HH:mm:ss");

String format = formatter.format(arg0);

LOG.info("Serialized Date='{}'", format);

arg1.writeString(format);

}

//and

// convert string in format MM/dd/yyyy HH:mm:ss to LocalDateTime

public class LocalDateTimeDeserializer extends JsonDeserializer<LocalDateTime> {

private static final Logger LOG = LoggerFactory.getLogger(LocalDateTimeDeserializer.class);

@Override

public LocalDateTime deserialize(JsonParser arg0, DeserializationContext arg1) throws IOException {

DateTimeFormatter formatter = DateTimeFormatter.ofPattern("MM/dd/yyyy HH:mm:ss");

LOG.info("Deserialized Date='{}'", arg0.getText());

return LocalDateTime.parse(arg0.getText(),formatter);

}

Create Kafka Producer and Consumer

If you want to understand deeply how to create Producer and Consumer with configuration, please the post Spring Boot Kafka Producer Consumer Configuration or You can also create Spring Boot Kafka Producer and Consumer without configuration, let check out the post Spring Boot Apache Kafka Example. Here I just introduce java source code for demonstration purposes.

Sending Kafka Messages to Topic

@Service
public class StaffSender {

    private static final Logger LOG = LoggerFactory.getLogger(StaffSender.class);

    @Autowired
    private KafkaTemplate<String, Staff> kafkaTemplate;

    @Value("${app.topic.staff}")
    private String topic;

    public void send(Staff data){
        LOG.info("sending JSON data='{}' to topic='{}'", data, topic);

        //create message object from JSON data object
        Message<Staff> message = MessageBuilder
                .withPayload(data)
                .setHeader(KafkaHeaders.TOPIC, topic)
                .build();

        kafkaTemplate.send(message);
    }
}

@Service

public class StaffSender {

private static final Logger LOG = LoggerFactory.getLogger(StaffSender.class);

@Autowired

private KafkaTemplate<String, Staff> kafkaTemplate;

@Value("${app.topic.staff}")

private String topic;

public void send(Staff data){

LOG.info("sending JSON data='{}' to topic='{}'", data, topic);

//create message object from JSON data object

Message<Staff> message = MessageBuilder

.withPayload(data)

.setHeader(KafkaHeaders.TOPIC, topic)

.build();

kafkaTemplate.send(message);

}

Here is the Spring Boot Kafka Producer configuration.

@Configuration
public class StaffSenderConfig {

    @Value("${spring.kafka.servers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return props;
    }

    @Bean
    public ProducerFactory<String, Staff> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, Staff> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}

@Configuration

public class StaffSenderConfig {

@Value("${spring.kafka.servers}")

private String bootstrapServers;

@Bean

public Map<String, Object> producerConfigs() {

Map<String, Object> props = new HashMap<>();

props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

return props;

}

@Bean

public ProducerFactory<String, Staff> producerFactory() {

return new DefaultKafkaProducerFactory<>(producerConfigs());

}

@Bean

public KafkaTemplate<String, Staff> kafkaTemplate() {

return new KafkaTemplate<>(producerFactory());

}

Receiving Kafka Messages from Topic

@Service
public class StaffReceiver {

    private static final Logger LOG = LoggerFactory.getLogger(StaffReceiver.class);

    @KafkaListener(topics = "${app.topic.staff}")
    public void receive(@Payload Staff data,
                        @Headers MessageHeaders headers) {
        LOG.info("received JSON data='{}'", data);

        headers.keySet().forEach(key -> {
            LOG.info("{}: {}", key, headers.get(key));
        });
    }

}

@Service

public class StaffReceiver {

private static final Logger LOG = LoggerFactory.getLogger(StaffReceiver.class);

@KafkaListener(topics = "${app.topic.staff}")

public void receive(@Payload Staff data,

@Headers MessageHeaders headers) {

LOG.info("received JSON data='{}'", data);

headers.keySet().forEach(key -> {

LOG.info("{}: {}", key, headers.get(key));

});

}

Here is the Spring Boot Kafka Consumer configuration:

@Configuration
@EnableKafka
public class StaffReceiverConfig {

    @Value("${spring.kafka.servers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "json");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return props;
    }

    @Bean
    public ConsumerFactory<String, Staff> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(
                consumerConfigs(),
                new StringDeserializer(),
                new JsonDeserializer<>(Staff.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Staff> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Staff> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

}

@Configuration

@EnableKafka

public class StaffReceiverConfig {

@Value("${spring.kafka.servers}")

private String bootstrapServers;

@Bean

public Map<String, Object> consumerConfigs() {

Map<String, Object> props = new HashMap<>();

props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

props.put(ConsumerConfig.GROUP_ID_CONFIG, "json");

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

return props;

}

@Bean

public ConsumerFactory<String, Staff> consumerFactory() {

return new DefaultKafkaConsumerFactory<>(

consumerConfigs(),

new StringDeserializer(),

new JsonDeserializer<>(Staff.class));

}

@Bean

public ConcurrentKafkaListenerContainerFactory<String, Staff> kafkaListenerContainerFactory() {

ConcurrentKafkaListenerContainerFactory<String, Staff> factory =

new ConcurrentKafkaListenerContainerFactory<>();

factory.setConsumerFactory(consumerFactory());

return factory;

}

Spring Boot Application

To demo our Spring Boot Kafka Json Serializer Example then we create a simple Spring Boot Application below.

@SpringBootApplication
public class MySpringKafkaApplication implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication.run(MySpringKafkaApplication.class, args);
    }

    @Autowired
    private StaffSender sender;

    @Override
    public void run(String... strings) throws Exception {
    	DateTimeFormatter formatter = DateTimeFormatter.ofPattern("MM/dd/yyyy HH:mm:ss");
        Staff staff = new Staff("JavabyCode", "United State", LocalDateTime.parse("07/15/2018 00:00:00", formatter));
        sender.send(staff);
    }
}

@SpringBootApplication

public class MySpringKafkaApplication implements CommandLineRunner {

public static void main(String[] args) {

SpringApplication.run(MySpringKafkaApplication.class, args);

}

@Autowired

private StaffSender sender;

@Override

public void run(String... strings) throws Exception {

DateTimeFormatter formatter = DateTimeFormatter.ofPattern("MM/dd/yyyy HH:mm:ss");

Staff staff = new Staff("JavabyCode", "United State", LocalDateTime.parse("07/15/2018 00:00:00", formatter));

sender.send(staff);

}

Running Spring Boot Application

Just run Spring Boot Application normally, it will print the output below

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::        (v2.2.2.RELEASE)

2020-02-10 15:49:14.524  INFO 20136 --- [           main] c.j.kafka.MySpringKafkaApplication       : Starting MySpringKafkaApplication on DESKTOP-BP6EDPD with PID 20136 (D:\Projects\spring-kafka-json-serializer-deserializer-example\target\classes started by Admin in D:\Projects\spring-kafka-json-serializer-deserializer-example)
2020-02-10 15:49:14.527 DEBUG 20136 --- [           main] c.j.kafka.MySpringKafkaApplication       : Running with Spring Boot v2.2.2.RELEASE, Spring v5.2.2.RELEASE
2020-02-10 15:49:14.527  INFO 20136 --- [           main] c.j.kafka.MySpringKafkaApplication       : No active profile set, falling back to default profiles: default
2020-02-10 15:49:15.688  INFO 20136 --- [           main] c.j.kafka.MySpringKafkaApplication       : Started MySpringKafkaApplication in 1.63 seconds (JVM running for 5.287)
2020-02-10 15:49:15.694  INFO 20136 --- [           main] c.javabycode.kafka.producer.StaffSender  : sending JSON data='Staff{name='JavabyCode', country='United State', birthday='2018-07-15T00:00'}' to topic='staff.t'
2020-02-10 15:49:15.894  INFO 20136 --- [           main] c.j.kafka.LocalDateTimeSerializer        : Serialized Date='07/15/2018 00:00:00'
2020-02-10 15:49:15.947  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.LocalDateTimeDeserializer      : Deserialized Date='07/15/2018 00:00:00'
2020-02-10 15:49:15.955  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver<span style="color: #ff0000;">         : received JSON data='Staff{name='JavabyCode', country='United State', birthday='2018-07-15T00:00'}'
</span>2020-02-10 15:49:15.955  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver         : kafka_offset: 12
2020-02-10 15:49:15.956  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver         : kafka_consumer: org.apache.kafka.clients.consumer.KafkaConsumer@13d9e0a5
2020-02-10 15:49:15.956  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver         : kafka_timestampType: CREATE_TIME
2020-02-10 15:49:15.956  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver         : kafka_receivedMessageKey: null
2020-02-10 15:49:15.956  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver         : kafka_receivedPartitionId: 0
2020-02-10 15:49:15.956  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver         : kafka_receivedTopic: staff.t
2020-02-10 15:49:15.956  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver         : kafka_receivedTimestamp: 1581324555901
2020-02-10 15:49:15.956  INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver         : kafka_groupId: json

. ____ _ __ _ _

/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \

( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \

\\/ ___)| |_)| | | | | || (_| | ) ) ) )

' |____| .__|_| |_|_| |_\__, | / / / /

=========|_|==============|___/=/_/_/_/

:: Spring Boot :: (v2.2.2.RELEASE)

2020-02-10 15:49:14.524 INFO 20136 --- [ main] c.j.kafka.MySpringKafkaApplication : Starting MySpringKafkaApplication on DESKTOP-BP6EDPD with PID 20136 (D:\Projects\spring-kafka-json-serializer-deserializer-example\target\classes started by Admin in D:\Projects\spring-kafka-json-serializer-deserializer-example)

2020-02-10 15:49:14.527 DEBUG 20136 --- [ main] c.j.kafka.MySpringKafkaApplication : Running with Spring Boot v2.2.2.RELEASE, Spring v5.2.2.RELEASE

2020-02-10 15:49:14.527 INFO 20136 --- [ main] c.j.kafka.MySpringKafkaApplication : No active profile set, falling back to default profiles: default

2020-02-10 15:49:15.688 INFO 20136 --- [ main] c.j.kafka.MySpringKafkaApplication : Started MySpringKafkaApplication in 1.63 seconds (JVM running for 5.287)

2020-02-10 15:49:15.694 INFO 20136 --- [ main] c.javabycode.kafka.producer.StaffSender : sending JSON data='Staff{name='JavabyCode', country='United State', birthday='2018-07-15T00:00'}' to topic='staff.t'

2020-02-10 15:49:15.894 INFO 20136 --- [ main] c.j.kafka.LocalDateTimeSerializer : Serialized Date='07/15/2018 00:00:00'

2020-02-10 15:49:15.947 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.LocalDateTimeDeserializer : Deserialized Date='07/15/2018 00:00:00'

2020-02-10 15:49:15.955 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver<span style="color: #ff0000;"> : received JSON data='Staff{name='JavabyCode', country='United State', birthday='2018-07-15T00:00'}'

</span>2020-02-10 15:49:15.955 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver : kafka_offset: 12

2020-02-10 15:49:15.956 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver : kafka_consumer: org.apache.kafka.clients.consumer.KafkaConsumer@13d9e0a5

2020-02-10 15:49:15.956 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver : kafka_timestampType: CREATE_TIME

2020-02-10 15:49:15.956 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver : kafka_receivedMessageKey: null

2020-02-10 15:49:15.956 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver : kafka_receivedPartitionId: 0

2020-02-10 15:49:15.956 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver : kafka_receivedTopic: staff.t

2020-02-10 15:49:15.956 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver : kafka_receivedTimestamp: 1581324555901

2020-02-10 15:49:15.956 INFO 20136 --- [ntainer#0-0-C-1] c.j.kafka.consumer.StaffReceiver : kafka_groupId: json

That’s all about Spring Boot Kafka Json Serializer Example. Now you can try to do your own practices and don’t forget to download the complete source code of Spring Boot Kafka Json Serializer Example below.

Download the complete source code

spring-kafka-json-serializer-deserializer-example.zip (324 downloads)

References

Spring Boot Kafka Producer Consumer Configuration
Spring Boot Apache Kafka Example
Spring Kafka Serialize Deserialize Documentation

Post Views: 4,066

0 Shares