In this guide, we'll walk through the steps to create a custom Plain Old Java Object (POJO) class, such as a Customer, and send it from a producer to a Kafka topic. This process involves creating the POJO, setting up the producer, and sending the object. Let's get started!
First, we need to define our custom object. In this example, we'll create a Customer class with some basic attributes.
public class Customer {
private String id;
private String name;
private String email;
// Default constructor
public Customer() {}
// Parameterized constructor
public Customer(String id, String name, String email) {
this.id = id;
this.name = name;
this.email = email;
}
// Getters and setters
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getEmail() {
return email;
}
public void setEmail(String email) {
this.email = email;
}
@Override
public String toString() {
return "Customer{" +
"id='" + id + '\'' +
", name='" + name + '\'' +
", email='" + email + '\'' +
'}';
}
}Next, we need to set up a Kafka producer that can send our Customer object to a Kafka topic. We'll use the Kafka producer API for this purpose.
First, add the necessary dependencies to your pom.xml:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>Then, configure the producer properties and create a producer instance:
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
public class CustomerProducer {
public static void main(String[] args) {
// Set up producer properties
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", "com.example.CustomerSerializer"); // Custom serializer for Customer
// Create the producer
Producer<String, Customer> producer = new KafkaProducer<>(props);
// Create a customer object
Customer customer = new Customer("1", "John Doe", "john.doe@example.com");
// Send the customer object to a Kafka topic
producer.send(new ProducerRecord<>("customer-topic", customer.getId(), customer));
// Close the producer
producer.close();
}
}To send our Customer object, we need a custom serializer. Here's how you can create one:
import org.apache.kafka.common.serialization.Serializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Map;
public class CustomerSerializer implements Serializer<Customer> {
@Override
public void configure(Map<String, ?> configs, boolean isKey) {}
@Override
public byte[] serialize(String topic, Customer data) {
byte[] retVal = null;
ObjectMapper objectMapper = new ObjectMapper();
try {
retVal = objectMapper.writeValueAsString(data).getBytes();
} catch (Exception e) {
e.printStackTrace();
}
return retVal;
}
@Override
public void close() {}
}By following these steps, you can create a custom POJO class and send it from a producer to a Kafka topic. This approach allows you to work with complex data structures in your Kafka-based applications. For more information on configuring Kafka for object serialization, check out our Configuring Kafka for Object Serialization guide.
Happy coding!