Currently I have code like this:
KafkaTemplate<String, String> kafkaTemplate;
List<Pet> myData;
for(Pet p: myData) {
String json = objectWriter.writeValueAsString(p)
kafkaTemplate.send(topic, json)
}
so each list item is sent one by one. How do i send the entire list at once?
So there is no direct way to send the bulk messages to kafka directly using KafkaTemplate or KafkaProducer. They don't any method that accept List of objects and send them individually to different partitions.
How do kafka producer send messages to kafka ?
Kafka producer create batch of records and then send these all records at once, for more information
The producer consists of a pool of buffer space that holds records that haven't yet been transmitted to the server as well as a background I/O thread that is responsible for turning these records into requests and transmitting them to the cluster.
The send() method is asynchronous. When called it adds the record to a buffer of pending record sends and immediately returns. This allows the producer to batch together individual records for efficiency.
Batching is one of the big drivers of efficiency, and to enable batching the Kafka producer will attempt to accumulate data in memory and to send out larger batches in a single request. The batching can be configured to accumulate no more than a fixed number of messages and to wait no longer than some fixed latency bound (say 64k or 10 ms). This allows the accumulation of more bytes to send, and few larger I/O operations on the servers. This buffering is configurable and gives a mechanism to trade off a small amount of additional latency for better throughput.
Since you are using spring-kafka you can send List<Objects> but here you are sending JSONArray of JSONObject instead of each JSONObject to topic partition
public KafkaTemplate<String, List<Object>> createTemplate() {
Map<String, Object> senderProps = senderProps();
ProducerFactory<Integer, String> pf =
new DefaultKafkaProducerFactory<String, List<Object>>(senderProps);
KafkaTemplate<String, List<Object>> template = new KafkaTemplate<>(pf);
return template;
}
public Map<String, Object> producerProps() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
return props;
}
KafkaTemplate<String, List<Object>> kafkaTemplate;
generally it's enough to set the property:
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
and increase the size of batch buffer with property:
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
According to your requirements.
NOTE:
Make sure that there are no flush() method calls in your code as it would discard all batch settings.
Using KafkaTemplate already answered by Deadpool there is another approach to just searlized object to byte array and send whole object.
This is not best approach as its overruled Kafka best practice to distribute and parallelize as much as possible . So in general distribute message and let producer to use pool buffer and partitions to parallelize. But some time we might need to use specific use........
//You can use any list of object i am just using String but could be enhance to any Object after all String is nothing but object..If you are using Pojo can be convert to JSON string and pass as List of String of json.
public byte[] searlizedByteArray(List<String> listObject) throws IOException {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
ObjectOutput out = null;
byte[] inByteArray = null;
try {
out = new ObjectOutputStream(bos);
out.writeObject(listObject);
out.flush();
inByteArray = bos.toByteArray();
} finally {
if (bos != null)
bos.close();
}
return inByteArray;
}
De-searlize byte[] array to List of object
public List<String> desearlizedByteArray(byte[] byteArray) throws IOException, ClassNotFoundException {
ByteArrayInputStream bis = new ByteArrayInputStream(byteArray);
ObjectInput in = null;
List<String> returnList=null;
try {
in = new ObjectInputStream(bis);
List<String> o = (List<String>) in.readObject();
for (String string : o) {
System.out.println("==="+o);
}
} finally {
try {
if (in != null) {
in.close();
}
} catch (IOException ex) {
// ignore close exception
}
}
return returnList;
}
Please note we are using VALUE_SERIALIZER_CLASS_CONFIG as ByteArraySerializer
public void publishMessage() throws Exception {
Properties properties = new Properties();
properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:8080");
properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
//You might to increase buffer memory and request size in case of large size,
properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "");
properties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "");
Producer<String, byte[]> producer = new org.apache.kafka.clients.producer.KafkaProducer<String, byte[]>(
properties);
try {
List asl=new ArrayList<String>();
asl.add("test1");
asl.add("test2");
byte[] byteArray = searlizedByteArray(asl);
ProducerRecord producerRecord = new ProducerRecord<String, byte[]>("testlist", null,
byteArray);
producer.send(producerRecord).get();
} finally {
if (producer != null) {
producer.flush();
producer.close();
}
}
}
Finally in consumer ByteArrayDeserializer will work
public void consumeMessage() {
Properties properties = new Properties();
properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9091");
properties.setProperty("key.deserializer", StringDeserializer.class.getName());
properties.setProperty("value.deserializer", ByteArrayDeserializer.class.getName());
properties.setProperty("group.id", "grp_consumer");
properties.put("auto.offset.reset", "earliest");
KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<String, byte[]>(properties);
consumer.subscribe(Arrays.asList("testlist"));
while (true) {
ConsumerRecords<String, byte[]> records = consumer.poll(100);
for (ConsumerRecord<String, byte[]> record : records) {
}
}
}