diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ExtSeqWithTimeWindowConsumer.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ExtSeqWithTimeWindowConsumer.java index d342c1a950..19595d9e95 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ExtSeqWithTimeWindowConsumer.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ExtSeqWithTimeWindowConsumer.java @@ -6,20 +6,21 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.serialization.LongDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; import java.time.Duration; import java.util.*; public class ExtSeqWithTimeWindowConsumer { - private static final long BUFFER_PERIOD_MS = 5000; + private static final long BUFFER_PERIOD_NS = 5000L * 1000000; // 5000 milliseconds converted to nanoseconds private static final Duration TIMEOUT_WAIT_FOR_MESSAGES = Duration.ofMillis(100); public static void main(String[] args) { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group"); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JacksonDeserializer.class.getName()); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put(Config.CONSUMER_VALUE_DESERIALIZER_SERIALIZED_CLASS, Message.class); @@ -32,7 +33,7 @@ public class ExtSeqWithTimeWindowConsumer { records.forEach(record -> { buffer.add(record.value()); }); - if (System.nanoTime() - lastProcessedTime > BUFFER_PERIOD_MS) { + if (System.nanoTime() - lastProcessedTime > BUFFER_PERIOD_NS) { processBuffer(buffer); lastProcessedTime = System.nanoTime(); } @@ -42,7 +43,7 @@ public class ExtSeqWithTimeWindowConsumer { private static void processBuffer(List buffer) { Collections.sort(buffer); buffer.forEach(message -> { - System.out.println("Processing message with Insert Position: " + message.getPartitionKey() + ", Message Id: " + message.getMessageId()); + System.out.println("Processing message with Partition key: " + message.getPartitionKey() + ", Application Identifier: " + message.getApplicationIdentifier()); }); buffer.clear(); } diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ExtSeqWithTimeWindowProducer.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ExtSeqWithTimeWindowProducer.java index d1480522e5..a20c569159 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ExtSeqWithTimeWindowProducer.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/ExtSeqWithTimeWindowProducer.java @@ -5,7 +5,7 @@ import com.baeldung.kafka.message.ordering.serialization.JacksonSerializer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.common.serialization.LongSerializer; import java.util.Properties; @@ -13,16 +13,14 @@ public class ExtSeqWithTimeWindowProducer { public static void main(String[] args) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JacksonSerializer.class.getName()); - - KafkaProducer producer = new KafkaProducer<>(props); - for (long insertPosition = 1; insertPosition <= 10 ; insertPosition++) { - long messageId = Message.getRandomMessageId(); - String key = "Key-" + insertPosition; - Message message = new Message(key, messageId); - producer.send(new ProducerRecord<>("multi_partition_topic", key, message)); - System.out.println("Insert Position: " + message.getPartitionKey() + ", Message Id: " + message.getMessageId()); + KafkaProducer producer = new KafkaProducer<>(props); + for (long partitionKey = 1; partitionKey <= 10 ; partitionKey++) { + long applicationIdentifier = Message.getRandomApplicationIdentifier(); + Message message = new Message(partitionKey, applicationIdentifier); + producer.send(new ProducerRecord<>("multi_partition_topic", partitionKey, message)); + System.out.println("Partition key: " + message.getPartitionKey() + ", Application Identifier: " + message.getApplicationIdentifier()); } producer.close(); System.out.println("ExternalSequencingProducer Completed."); diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionConsumer.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionConsumer.java index 4471070f0f..c37c645d31 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionConsumer.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/MultiPartitionConsumer.java @@ -6,6 +6,7 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.serialization.LongDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; import java.time.Duration; @@ -18,7 +19,7 @@ public class MultiPartitionConsumer { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group"); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JacksonDeserializer.class.getName()); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put(Config.CONSUMER_VALUE_DESERIALIZER_SERIALIZED_CLASS, Message.class); @@ -29,7 +30,7 @@ public class MultiPartitionConsumer { records.forEach(record -> { Message message = record.value(); if (message != null) { - System.out.println("Process message with Insert Position: " + message.getPartitionKey() + ", Message Id: " + message.getMessageId()); + System.out.println("Process message with Partition key: " + message.getPartitionKey() + ", Application Identifier: " + message.getApplicationIdentifier()); } }); } 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 04e3dcce0a..81cc5c6af0 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 @@ -5,7 +5,7 @@ import com.baeldung.kafka.message.ordering.serialization.JacksonSerializer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.common.serialization.LongSerializer; import java.util.Properties; @@ -13,16 +13,14 @@ public class MultiPartitionProducer { public static void main(String[] args) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JacksonSerializer.class.getName()); - - KafkaProducer producer = new KafkaProducer<>(props); - for (long insertPosition = 1; insertPosition <= 10 ; insertPosition++) { - long messageId = Message.getRandomMessageId(); - String key = "Key-" + insertPosition; - Message message = new Message(key, messageId); - producer.send(new ProducerRecord<>("multi_partition_topic", key, message)); - System.out.println("Insert Position: " + message.getPartitionKey() + ", Message Id: " + message.getMessageId()); + KafkaProducer producer = new KafkaProducer<>(props); + for (long partitionKey = 1; partitionKey <= 10 ; partitionKey++) { + long applicationIdentifier = Message.getRandomApplicationIdentifier(); + Message message = new Message(partitionKey, applicationIdentifier); + producer.send(new ProducerRecord<>("multi_partition_topic", partitionKey, message)); + System.out.println("Partition Key: " + message.getPartitionKey() + ", Application Identifier: " + message.getApplicationIdentifier()); } producer.close(); System.out.println("SinglePartitionProducer Completed."); diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/SinglePartitionConsumer.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/SinglePartitionConsumer.java index b47e4ca3b0..9f44cd78c6 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/SinglePartitionConsumer.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/SinglePartitionConsumer.java @@ -6,6 +6,7 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.common.serialization.LongDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; import java.time.Duration; @@ -19,17 +20,17 @@ public class SinglePartitionConsumer { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group"); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JacksonDeserializer.class.getName()); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put(Config.CONSUMER_VALUE_DESERIALIZER_SERIALIZED_CLASS, Message.class); - Consumer consumer = new KafkaConsumer<>(props); + Consumer consumer = new KafkaConsumer<>(props); consumer.subscribe(Collections.singletonList("single_partition_topic")); while (true) { - ConsumerRecords records = consumer.poll(TIMEOUT_WAIT_FOR_MESSAGES); + ConsumerRecords records = consumer.poll(TIMEOUT_WAIT_FOR_MESSAGES); records.forEach(record -> { Message message = record.value(); - System.out.println("Process message with Insert Position: " + message.getPartitionKey() + ", Message Id: " + message.getMessageId()); + System.out.println("Process message with Partition Key: " + message.getPartitionKey() + ", Application Identifier: " + message.getApplicationIdentifier()); }); } } diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/SinglePartitionProducer.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/SinglePartitionProducer.java index d669a0fd69..efa6e5b93d 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/SinglePartitionProducer.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/SinglePartitionProducer.java @@ -5,7 +5,7 @@ import com.baeldung.kafka.message.ordering.serialization.JacksonSerializer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.common.serialization.LongSerializer; import java.util.Properties; @@ -13,16 +13,14 @@ public class SinglePartitionProducer { public static void main(String[] args) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JacksonSerializer.class.getName()); - - KafkaProducer producer = new KafkaProducer<>(props); - for (long insertPosition = 1; insertPosition <= 10 ; insertPosition++) { - long messageId = Message.getRandomMessageId(); - String key = "Key-" + insertPosition; - Message message = new Message(key, messageId); - producer.send(new ProducerRecord<>("single_partition_topic", key, message)); - System.out.println("Insert Position: " + message.getPartitionKey() + ", Message Id: " + message.getMessageId()); + KafkaProducer producer = new KafkaProducer<>(props); + for (long partitionKey = 1; partitionKey <= 10 ; partitionKey++) { + long applicationIdentifier = Message.getRandomApplicationIdentifier(); + Message message = new Message(partitionKey, applicationIdentifier); + producer.send(new ProducerRecord<>("single_partition_topic", partitionKey, message)); + System.out.println("Partition key: " + message.getPartitionKey() + ", Application Identifier: " + message.getApplicationIdentifier()); } producer.close(); System.out.println("SinglePartitionProducer Completed."); diff --git a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/payload/Message.java b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/payload/Message.java index de1e5135da..734ecba53d 100644 --- a/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/payload/Message.java +++ b/apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/payload/Message.java @@ -5,30 +5,30 @@ import java.util.Random; import java.util.concurrent.ThreadLocalRandom; public class Message implements Comparable { - private String partitionKey; - private long messageId; + private long partitionKey; + private long applicationIdentifier; public Message(){ } //Required for Kafka Serialization and Deserialization - public Message(String partitionKey, long messageId) { + public Message(long partitionKey, long applicationIdentifier) { this.partitionKey = partitionKey; - this.messageId = messageId; + this.applicationIdentifier = applicationIdentifier; } - public String getPartitionKey() { + public long getPartitionKey() { return partitionKey; } - public long getMessageId() { - return messageId; + public long getApplicationIdentifier() { + return applicationIdentifier; } @Override public int compareTo(Message other) { - return Long.compare(this.messageId, other.messageId); + return Long.compare(this.partitionKey, other.partitionKey); } @Override @@ -40,10 +40,10 @@ public class Message implements Comparable { return false; } Message message = (Message) obj; - return this.messageId == message.getMessageId() && Objects.equals(this.partitionKey, message.getPartitionKey()); + return this.applicationIdentifier == message.getApplicationIdentifier() && Objects.equals(this.partitionKey, message.getPartitionKey()); } - public static long getRandomMessageId() { + public static long getRandomApplicationIdentifier() { Random rand = new Random(); return ThreadLocalRandom.current().nextInt(1000); } diff --git a/apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/MultiplePartitionTest.java b/apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/MultiplePartitionTest.java index aed5f30e9d..ef0a881999 100644 --- a/apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/MultiplePartitionTest.java +++ b/apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/MultiplePartitionTest.java @@ -3,7 +3,6 @@ package com.baeldung.kafka.message.ordering; import com.baeldung.kafka.message.ordering.payload.Message; import com.baeldung.kafka.message.ordering.serialization.JacksonDeserializer; import com.baeldung.kafka.message.ordering.serialization.JacksonSerializer; -import lombok.var; import org.apache.kafka.clients.admin.*; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecords; @@ -13,9 +12,8 @@ import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.KafkaFuture; -import org.apache.kafka.common.PartitionInfo; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.common.serialization.LongDeserializer; +import org.apache.kafka.common.serialization.LongSerializer; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -36,8 +34,8 @@ public class MultiplePartitionTest { private static int PARTITIONS = 5; private static short REPLICATION_FACTOR = 1; private static Admin admin; - private static KafkaProducer producer; - private static KafkaConsumer consumer; + private static KafkaProducer producer; + private static KafkaConsumer consumer; private static final Duration TIMEOUT_WAIT_FOR_MESSAGES = Duration.ofMillis(5000); @Container private static final KafkaContainer KAFKA_CONTAINER = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest")); @@ -51,12 +49,12 @@ public class MultiplePartitionTest { Properties producerProperties = new Properties(); producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers()); - producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName()); producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JacksonSerializer.class.getName()); Properties consumerProperties = new Properties(); consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers()); - consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName()); consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JacksonDeserializer.class.getName()); consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); consumerProperties.put(Config.CONSUMER_VALUE_DESERIALIZER_SERIALIZED_CLASS, Message.class); @@ -87,11 +85,10 @@ public class MultiplePartitionTest { void givenMultiplePartitions_whenPublishedToKafkaAndConsumed_thenCheckForMessageOrder() throws ExecutionException, InterruptedException { List sentMessageList = new ArrayList<>(); List receivedMessageList = new ArrayList<>(); - for (long insertPosition = 1; insertPosition <= 10 ; insertPosition++) { - long messageId = Message.getRandomMessageId(); - String key = "Key-" + insertPosition; - Message message = new Message(key, messageId); - Future future = producer.send(new ProducerRecord<>(TOPIC, key, message)); + for (long partitionKey = 1; partitionKey <= 10 ; partitionKey++) { + long applicationIdentifier = Message.getRandomApplicationIdentifier(); + Message message = new Message(partitionKey, applicationIdentifier); + Future future = producer.send(new ProducerRecord<>(TOPIC, partitionKey, message)); sentMessageList.add(message); RecordMetadata metadata = future.get(); System.out.println("Partition : " + metadata.partition()); @@ -99,7 +96,7 @@ public class MultiplePartitionTest { boolean isOrderMaintained = true; consumer.subscribe(Collections.singletonList(TOPIC)); - ConsumerRecords records = consumer.poll(TIMEOUT_WAIT_FOR_MESSAGES); + ConsumerRecords records = consumer.poll(TIMEOUT_WAIT_FOR_MESSAGES); records.forEach(record -> { Message message = record.value(); receivedMessageList.add(message); diff --git a/apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/SinglePartitionTest.java b/apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/SinglePartitionTest.java index 5751c8d0e0..350a28e7c1 100644 --- a/apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/SinglePartitionTest.java +++ b/apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/SinglePartitionTest.java @@ -15,8 +15,8 @@ import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.KafkaFuture; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.common.serialization.LongDeserializer; +import org.apache.kafka.common.serialization.LongSerializer; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -36,8 +36,8 @@ public class SinglePartitionTest { private static int PARTITIONS = 1; private static short REPLICATION_FACTOR = 1; private static Admin admin; - private static KafkaProducer producer; - private static KafkaConsumer consumer; + private static KafkaProducer producer; + private static KafkaConsumer consumer; private static final Duration TIMEOUT_WAIT_FOR_MESSAGES = Duration.ofMillis(5000); @@ -53,12 +53,12 @@ public class SinglePartitionTest { Properties producerProperties = new Properties(); producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers()); - producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName()); producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JacksonSerializer.class.getName()); Properties consumerProperties = new Properties(); consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers()); - consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName()); consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JacksonDeserializer.class.getName()); consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); consumerProperties.put(Config.CONSUMER_VALUE_DESERIALIZER_SERIALIZED_CLASS, Message.class); @@ -89,11 +89,10 @@ public class SinglePartitionTest { void givenASinglePartition_whenPublishedToKafkaAndConsumed_thenCheckForMessageOrder() throws ExecutionException, InterruptedException { List sentMessageList = new ArrayList<>(); List receivedMessageList = new ArrayList<>(); - for (long insertPosition = 1; insertPosition <= 10 ; insertPosition++) { - long messageId = Message.getRandomMessageId(); - String key = "Key-" + insertPosition; - Message message = new Message(key, messageId); - ProducerRecord producerRecord = new ProducerRecord<>(TOPIC, key, message); + for (long partitionKey = 1; partitionKey <= 10 ; partitionKey++) { + long applicationIdentifier = Message.getRandomApplicationIdentifier(); + Message message = new Message(partitionKey, applicationIdentifier); + ProducerRecord producerRecord = new ProducerRecord<>(TOPIC, partitionKey, message); Future future = producer.send(producerRecord); sentMessageList.add(message); RecordMetadata metadata = future.get(); @@ -101,7 +100,7 @@ public class SinglePartitionTest { } consumer.subscribe(Collections.singletonList(TOPIC)); - ConsumerRecords records = consumer.poll(TIMEOUT_WAIT_FOR_MESSAGES); + ConsumerRecords records = consumer.poll(TIMEOUT_WAIT_FOR_MESSAGES); records.forEach(record -> { Message message = record.value(); receivedMessageList.add(message);