From f8ea1db5ccd727c2e06912a04eac3b9de4088341 Mon Sep 17 00:00:00 2001 From: Amol Gote Date: Sun, 15 Oct 2023 15:58:27 -0400 Subject: [PATCH 1/8] Update apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java Co-authored-by: Liam Williams --- .../baeldung/kafka/message/ordering/ProducerConfigurations.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java index 60ae5fc2f8..566605973c 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java @@ -12,7 +12,7 @@ public class ProducerConfigurations { props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("max.in.flight.requests.per.connection", "1"); - props.put("batch.size", "16384"); + props.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384"); props.put(ProducerConfig.LINGER_MS_CONFIG, "5"); KafkaProducer producer = new KafkaProducer<>(props); From 463e45d025682fbd4eecbb73165c460ef9f7ea03 Mon Sep 17 00:00:00 2001 From: Amol Gote Date: Sun, 15 Oct 2023 15:58:33 -0400 Subject: [PATCH 2/8] Update apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java Co-authored-by: Liam Williams --- .../baeldung/kafka/message/ordering/ProducerConfigurations.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java index 566605973c..c9cef04bbc 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java @@ -11,7 +11,7 @@ public class ProducerConfigurations { props.put("bootstrap.servers", "localhost:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); - props.put("max.in.flight.requests.per.connection", "1"); + props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1"); props.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384"); props.put(ProducerConfig.LINGER_MS_CONFIG, "5"); KafkaProducer producer = new KafkaProducer<>(props); From 836c977fdea79eda47f2e5989709d73f40e5f593 Mon Sep 17 00:00:00 2001 From: Amol Gote Date: Sun, 15 Oct 2023 15:58:43 -0400 Subject: [PATCH 3/8] Update apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java Co-authored-by: Liam Williams --- .../baeldung/kafka/message/ordering/ProducerConfigurations.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java index c9cef04bbc..81d91693fe 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java @@ -8,7 +8,7 @@ import java.util.Properties; public class ProducerConfigurations { public static void main(String[] args) { Properties props = new Properties(); - props.put("bootstrap.servers", "localhost:9092"); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1"); From 540da8c7795ac450793347b927e6c6a800c06c25 Mon Sep 17 00:00:00 2001 From: Amol Gote Date: Sun, 15 Oct 2023 15:58:57 -0400 Subject: [PATCH 4/8] Update apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonSerializer.java Co-authored-by: Liam Williams --- .../message/ordering/serialization/JacksonSerializer.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonSerializer.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonSerializer.java index fa9d25dd85..2d7432cc7b 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonSerializer.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonSerializer.java @@ -3,6 +3,10 @@ package com.baeldung.kafka.message.ordering.serialization; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.kafka.common.serialization.Serializer; +/** + * Configured via {@link org.apache.kafka.clients.producer.ProducerConfig#VALUE_SERIALIZER_CLASS_CONFIG} + */ +@SuppressWarnings("unused") public class JacksonSerializer implements Serializer { private final ObjectMapper objectMapper = new ObjectMapper(); From c3af2facae6cdc5b3a27607eec2d079f1a670d9f Mon Sep 17 00:00:00 2001 From: Amol Gote Date: Sun, 15 Oct 2023 15:59:07 -0400 Subject: [PATCH 5/8] Update apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonDeserializer.java Co-authored-by: Liam Williams --- .../message/ordering/serialization/JacksonDeserializer.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonDeserializer.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonDeserializer.java index 34aa181fcb..f3c8aaf60f 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonDeserializer.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/serialization/JacksonDeserializer.java @@ -4,6 +4,10 @@ import org.apache.kafka.common.serialization.Deserializer; import java.util.Map; +/** + * Configured via {@link org.apache.kafka.clients.consumer.ConsumerConfig#VALUE_DESERIALIZER_CLASS_CONFIG} + */ +@SuppressWarnings("unused") public class JacksonDeserializer implements Deserializer { private final ObjectMapper objectMapper = new ObjectMapper(); private Class tClass; From d31cf69b4d5d2ca8d88cbebf6aee72b84dba96b3 Mon Sep 17 00:00:00 2001 From: Amol Gote Date: Sun, 15 Oct 2023 15:59:38 -0400 Subject: [PATCH 6/8] Update apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java Co-authored-by: Liam Williams --- .../baeldung/kafka/message/ordering/MultiPartitionProducer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java index 8b2a49b2b5..0fd20c2ab1 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java @@ -11,7 +11,7 @@ public class MultiPartitionProducer { Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); - props.put("value.serializer", "com.baeldung.kafka.message.ordering.serialization.JacksonSerializer"); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.baeldung.kafka.message.ordering.serialization.JacksonSerializer"); KafkaProducer producer = new KafkaProducer<>(props); for (long insertPosition = 1; insertPosition <= 10 ; insertPosition++) { From 857da48e742306babf0579436aa2baf0c3ea5a0c Mon Sep 17 00:00:00 2001 From: Amol Gote Date: Sun, 15 Oct 2023 16:00:04 -0400 Subject: [PATCH 7/8] Update apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java Co-authored-by: Liam Williams --- .../baeldung/kafka/message/ordering/ProducerConfigurations.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java index 81d91693fe..ae3f2839c8 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ProducerConfigurations.java @@ -10,7 +10,7 @@ public class ProducerConfigurations { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); - props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1"); props.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384"); props.put(ProducerConfig.LINGER_MS_CONFIG, "5"); From e35327cade00d1be3bedc513e416dc9de28f177a Mon Sep 17 00:00:00 2001 From: Amol Gote Date: Sun, 15 Oct 2023 16:00:49 -0400 Subject: [PATCH 8/8] Update apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java Co-authored-by: Liam Williams --- .../baeldung/kafka/message/ordering/MultiPartitionProducer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java index 0fd20c2ab1..04ffdd3336 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionProducer.java @@ -10,7 +10,7 @@ public class MultiPartitionProducer { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); - props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.baeldung.kafka.message.ordering.serialization.JacksonSerializer"); KafkaProducer producer = new KafkaProducer<>(props);