iSharkFly-Docs/java-tutorials
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fixed unit test case failures

Browse Source
This commit is contained in:
Amol Gote 2023-11-04 18:09:21 -04:00
parent 2cecae1dfb
commit 2a6e561f76
3 changed files with 12 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
apache-kafka-2/src/main/java/com/baeldung/kafka/message/ordering/payload/UserEvent.java
View File

@ -42,15 +42,19 @@ public class UserEvent implements Comparable<UserEvent> {
@Override @Override
public boolean equals(Object obj) { public boolean equals(Object obj) {
if (obj == this) { if (this == obj) {
return true; return true;
} }
if (!(obj instanceof UserEvent)) { if (!(obj instanceof UserEvent)) {
return false; return false;
} }
UserEvent userEvent = (UserEvent) obj; UserEvent userEvent = (UserEvent) obj;
return Objects.equals(this.userEventId, userEvent.getUserEventId()) return this.globalSequenceNumber == userEvent.globalSequenceNumber;
&& userEvent.getEventNanoTime() == this.eventNanoTime; }
@Override
public int hashCode() {
return Objects.hash(globalSequenceNumber);
} }
} }

6
apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/MultiplePartitionIntegrationTest.java
View File

@ -71,16 +71,16 @@ public class MultiplePartitionIntegrationTest {
void givenMultiplePartitions_whenPublishedToKafkaAndConsumed_thenCheckForMessageOrder() throws ExecutionException, InterruptedException { void givenMultiplePartitions_whenPublishedToKafkaAndConsumed_thenCheckForMessageOrder() throws ExecutionException, InterruptedException {
List<UserEvent> sentUserEventList = new ArrayList<>(); List<UserEvent> sentUserEventList = new ArrayList<>();
List<UserEvent> receivedUserEventList = new ArrayList<>(); List<UserEvent> receivedUserEventList = new ArrayList<>();
for (long count = 1; count <= 10 ; count++) { for (long sequenceNumber = 1; sequenceNumber <= 10; sequenceNumber++) {
UserEvent userEvent = new UserEvent(UUID.randomUUID().toString()); UserEvent userEvent = new UserEvent(UUID.randomUUID().toString());
userEvent.setGlobalSequenceNumber(sequenceNumber);
userEvent.setEventNanoTime(System.nanoTime()); userEvent.setEventNanoTime(System.nanoTime());
Future<RecordMetadata> future = producer.send(new ProducerRecord<>(Config.MULTI_PARTITION_TOPIC, count, userEvent)); Future<RecordMetadata> future = producer.send(new ProducerRecord<>(Config.MULTI_PARTITION_TOPIC, sequenceNumber, userEvent));
sentUserEventList.add(userEvent); sentUserEventList.add(userEvent);
RecordMetadata metadata = future.get(); RecordMetadata metadata = future.get();
System.out.println("User Event ID: " + userEvent.getUserEventId() + ", Partition : " + metadata.partition()); System.out.println("User Event ID: " + userEvent.getUserEventId() + ", Partition : " + metadata.partition());
} }
boolean isOrderMaintained = true;
consumer.subscribe(Collections.singletonList(Config.MULTI_PARTITION_TOPIC)); consumer.subscribe(Collections.singletonList(Config.MULTI_PARTITION_TOPIC));
ConsumerRecords<Long, UserEvent> records = consumer.poll(TIMEOUT_WAIT_FOR_MESSAGES); ConsumerRecords<Long, UserEvent> records = consumer.poll(TIMEOUT_WAIT_FOR_MESSAGES);
records.forEach(record -> { records.forEach(record -> {

3
apache-kafka-2/src/test/java/com/baeldung/kafka/message/ordering/SinglePartitionIntegrationTest.java
View File

@ -78,8 +78,9 @@ public class SinglePartitionIntegrationTest {
void givenASinglePartition_whenPublishedToKafkaAndConsumed_thenCheckForMessageOrder() throws ExecutionException, InterruptedException { void givenASinglePartition_whenPublishedToKafkaAndConsumed_thenCheckForMessageOrder() throws ExecutionException, InterruptedException {
List<UserEvent> sentUserEventList = new ArrayList<>(); List<UserEvent> sentUserEventList = new ArrayList<>();
List<UserEvent> receivedUserEventList = new ArrayList<>(); List<UserEvent> receivedUserEventList = new ArrayList<>();
for (long count = 1; count <= 10; count++) { for (long sequenceNumber = 1; sequenceNumber <= 10; sequenceNumber++) {
UserEvent userEvent = new UserEvent(UUID.randomUUID().toString()); UserEvent userEvent = new UserEvent(UUID.randomUUID().toString());
userEvent.setGlobalSequenceNumber(sequenceNumber);
userEvent.setEventNanoTime(System.nanoTime()); userEvent.setEventNanoTime(System.nanoTime());
ProducerRecord<Long, UserEvent> producerRecord = new ProducerRecord<>(Config.SINGLE_PARTITION_TOPIC, userEvent); ProducerRecord<Long, UserEvent> producerRecord = new ProducerRecord<>(Config.SINGLE_PARTITION_TOPIC, userEvent);
Future<RecordMetadata> future = producer.send(producerRecord); Future<RecordMetadata> future = producer.send(producerRecord);