Skip to content

Commit 97c8c6b

Browse files
sjhajhariasjhajharia
authored
KAFKA-19733 Fix arguments to assertEquals() in clients module (#20586)
The given PR mostly fixes the order of arguments in `assertEquals()` for the Clients module. Some minor cleanups were included with the same too. Reviewers: Chia-Ping Tsai <[email protected]>
1 parent 14917ae commit 97c8c6b

33 files changed

+215
-251
lines changed

‎clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIdExpirationTest.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -204,10 +204,10 @@ void testDynamicProducerIdExpirationMs(ClusterInstance cluster) throws Interrupt
204204
// Update the producer ID expiration ms to a very high value.
205205
admin.incrementalAlterConfigs(producerIdExpirationConfig("100000"));
206206

207-
cluster.brokers().values().forEach(broker -> {
207+
cluster.brokers().values().forEach(broker ->
208208
TestUtils.waitUntilTrue(() -> broker.logManager().producerStateManagerConfig().producerIdExpirationMs() == 100000,
209-
() -> "Configuration was not updated.", DEFAULT_MAX_WAIT_MS, 100);
210-
});
209+
() -> "Configuration was not updated.", DEFAULT_MAX_WAIT_MS, 100)
210+
);
211211
// Send more records to send producer ID back to brokers.
212212
producer.send(new ProducerRecord<>(topic1, 0, null, "key".getBytes(), "value".getBytes()));
213213
producer.flush();
@@ -226,10 +226,10 @@ void testDynamicProducerIdExpirationMs(ClusterInstance cluster) throws Interrupt
226226
kafkaBroker.awaitShutdown();
227227
kafkaBroker.startup();
228228
cluster.waitForReadyBrokers();
229-
cluster.brokers().values().forEach(broker -> {
229+
cluster.brokers().values().forEach(broker ->
230230
TestUtils.waitUntilTrue(() -> broker.logManager().producerStateManagerConfig().producerIdExpirationMs() == 100,
231-
() -> "Configuration was not updated.", DEFAULT_MAX_WAIT_MS, 100);
232-
});
231+
() -> "Configuration was not updated.", DEFAULT_MAX_WAIT_MS, 100)
232+
);
233233

234234
// Ensure producer ID expires quickly again.
235235
waitProducerIdExpire(admin);

‎clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerSendWhileDeletionTest.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -184,9 +184,8 @@ public void testSendWhileTopicGetRecreated() {
184184
try (var producer = createProducer()) {
185185
for (int i = 1; i <= numRecords; i++) {
186186
producer.send(new ProducerRecord<>(topic, null, ("value" + i).getBytes()),
187-
(metadata, exception) -> {
188-
numAcks.incrementAndGet();
189-
});
187+
(metadata, exception) -> numAcks.incrementAndGet()
188+
);
190189
}
191190
producer.flush();
192191
}

‎clients/src/test/java/org/apache/kafka/clients/ClusterConnectionStatesTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -186,7 +186,7 @@ public void testAuthorizationFailed() {
186186

187187
connectionStates.authenticationFailed(nodeId1, time.milliseconds(), new AuthenticationException("No path to CA for certificate!"));
188188
time.sleep(1000);
189-
assertEquals(connectionStates.connectionState(nodeId1), ConnectionState.AUTHENTICATION_FAILED);
189+
assertEquals(ConnectionState.AUTHENTICATION_FAILED, connectionStates.connectionState(nodeId1));
190190
assertNotNull(connectionStates.authenticationException(nodeId1));
191191
assertFalse(connectionStates.hasReadyNodes(time.milliseconds()));
192192
assertFalse(connectionStates.canConnect(nodeId1, time.milliseconds()));
@@ -210,7 +210,7 @@ public void testRemoveNode() {
210210
connectionStates.remove(nodeId1);
211211
assertTrue(connectionStates.canConnect(nodeId1, time.milliseconds()));
212212
assertFalse(connectionStates.isBlackedOut(nodeId1, time.milliseconds()));
213-
assertEquals(connectionStates.connectionDelay(nodeId1, time.milliseconds()), 0L);
213+
assertEquals(0L, connectionStates.connectionDelay(nodeId1, time.milliseconds()));
214214
}
215215

216216
@Test

‎clients/src/test/java/org/apache/kafka/clients/MetadataTest.java‎

Lines changed: 37 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -361,28 +361,28 @@ public void testUpdateLastEpoch() {
361361
// Metadata with newer epoch is handled
362362
metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 10);
363363
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 1L);
364-
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 10));
364+
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(10, leaderAndEpoch.intValue()));
365365

366366
// Don't update to an older one
367367
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 1));
368-
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 10));
368+
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(10, leaderAndEpoch.intValue()));
369369

370370
// Don't cause update if it's the same one
371371
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 10));
372-
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 10));
372+
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(10, leaderAndEpoch.intValue()));
373373

374374
// Update if we see newer epoch
375375
assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 12));
376-
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 12));
376+
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(12, leaderAndEpoch.intValue()));
377377

378378
metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 12);
379379
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 2L);
380-
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 12));
380+
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(12, leaderAndEpoch.intValue()));
381381

382382
// Don't overwrite metadata with older epoch
383383
metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 11);
384384
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 3L);
385-
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 12));
385+
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(12, leaderAndEpoch.intValue()));
386386
}
387387

388388
@Test
@@ -465,7 +465,7 @@ public void testRejectOldMetadata() {
465465
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 10L);
466466
assertNotNull(metadata.fetch().partition(tp));
467467
assertTrue(metadata.lastSeenLeaderEpoch(tp).isPresent());
468-
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
468+
assertEquals(100, metadata.lastSeenLeaderEpoch(tp).get().longValue());
469469
}
470470

471471
// Fake an empty ISR, but with an older epoch, should reject it
@@ -475,8 +475,8 @@ public void testRejectOldMetadata() {
475475
new MetadataResponse.PartitionMetadata(error, partition, leader,
476476
leaderEpoch, replicas, Collections.emptyList(), offlineReplicas), ApiKeys.METADATA.latestVersion(), Collections.emptyMap());
477477
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 20L);
478-
assertEquals(metadata.fetch().partition(tp).inSyncReplicas().length, 1);
479-
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
478+
assertEquals(1, metadata.fetch().partition(tp).inSyncReplicas().length);
479+
assertEquals(100, metadata.lastSeenLeaderEpoch(tp).get().longValue());
480480
}
481481

482482
// Fake an empty ISR, with same epoch, accept it
@@ -486,24 +486,24 @@ public void testRejectOldMetadata() {
486486
new MetadataResponse.PartitionMetadata(error, partition, leader,
487487
leaderEpoch, replicas, Collections.emptyList(), offlineReplicas), ApiKeys.METADATA.latestVersion(), Collections.emptyMap());
488488
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 20L);
489-
assertEquals(metadata.fetch().partition(tp).inSyncReplicas().length, 0);
490-
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
489+
assertEquals(0, metadata.fetch().partition(tp).inSyncReplicas().length);
490+
assertEquals(100, metadata.lastSeenLeaderEpoch(tp).get().longValue());
491491
}
492492

493493
// Empty metadata response, should not keep old partition but should keep the last-seen epoch
494494
{
495495
MetadataResponse metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.emptyMap());
496496
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 20L);
497497
assertNull(metadata.fetch().partition(tp));
498-
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
498+
assertEquals(100, metadata.lastSeenLeaderEpoch(tp).get().longValue());
499499
}
500500

501501
// Back in the metadata, with old epoch, should not get added
502502
{
503503
MetadataResponse metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 99);
504504
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 10L);
505505
assertNull(metadata.fetch().partition(tp));
506-
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
506+
assertEquals(100, metadata.lastSeenLeaderEpoch(tp).get().longValue());
507507
}
508508
}
509509

@@ -522,31 +522,31 @@ public void testOutOfBandEpochUpdate() {
522522
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 10L);
523523
assertNotNull(metadata.fetch().partition(tp));
524524
assertTrue(metadata.lastSeenLeaderEpoch(tp).isPresent());
525-
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
525+
assertEquals(100, metadata.lastSeenLeaderEpoch(tp).get().longValue());
526526

527527
// Simulate a leader epoch from another response, like a fetch response or list offsets
528528
assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 101));
529529

530530
// Cache of partition stays, but current partition info is not available since it's stale
531531
assertNotNull(metadata.fetch().partition(tp));
532-
assertEquals(Objects.requireNonNull(metadata.fetch().partitionCountForTopic("topic-1")).longValue(), 5);
532+
assertEquals(5, Objects.requireNonNull(metadata.fetch().partitionCountForTopic("topic-1")).longValue());
533533
assertFalse(metadata.partitionMetadataIfCurrent(tp).isPresent());
534-
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 101);
534+
assertEquals(101, metadata.lastSeenLeaderEpoch(tp).get().longValue());
535535

536536
// Metadata with older epoch is rejected, metadata state is unchanged
537537
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 20L);
538538
assertNotNull(metadata.fetch().partition(tp));
539-
assertEquals(Objects.requireNonNull(metadata.fetch().partitionCountForTopic("topic-1")).longValue(), 5);
539+
assertEquals(5, Objects.requireNonNull(metadata.fetch().partitionCountForTopic("topic-1")).longValue());
540540
assertFalse(metadata.partitionMetadataIfCurrent(tp).isPresent());
541-
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 101);
541+
assertEquals(101, metadata.lastSeenLeaderEpoch(tp).get().longValue());
542542

543543
// Metadata with equal or newer epoch is accepted
544544
metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 101);
545545
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 30L);
546546
assertNotNull(metadata.fetch().partition(tp));
547-
assertEquals(Objects.requireNonNull(metadata.fetch().partitionCountForTopic("topic-1")).longValue(), 5);
547+
assertEquals(5, Objects.requireNonNull(metadata.fetch().partitionCountForTopic("topic-1")).longValue());
548548
assertTrue(metadata.partitionMetadataIfCurrent(tp).isPresent());
549-
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 101);
549+
assertEquals(101, metadata.lastSeenLeaderEpoch(tp).get().longValue());
550550
}
551551

552552
@Test
@@ -585,18 +585,18 @@ public void testClusterCopy() {
585585
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 0L);
586586

587587
Cluster cluster = metadata.fetch();
588-
assertEquals(cluster.clusterResource().clusterId(), "dummy");
589-
assertEquals(cluster.nodes().size(), 4);
588+
assertEquals("dummy", cluster.clusterResource().clusterId());
589+
assertEquals(4, cluster.nodes().size());
590590

591591
// topic counts
592592
assertEquals(cluster.invalidTopics(), Collections.singleton("topic3"));
593593
assertEquals(cluster.unauthorizedTopics(), Collections.singleton("topic4"));
594-
assertEquals(cluster.topics().size(), 3);
594+
assertEquals(3, cluster.topics().size());
595595
assertEquals(cluster.internalTopics(), Collections.singleton(Topic.GROUP_METADATA_TOPIC_NAME));
596596

597597
// partition counts
598-
assertEquals(cluster.partitionsForTopic("topic1").size(), 2);
599-
assertEquals(cluster.partitionsForTopic("topic2").size(), 3);
598+
assertEquals(2, cluster.partitionsForTopic("topic1").size());
599+
assertEquals(3, cluster.partitionsForTopic("topic2").size());
600600

601601
// Sentinel instances
602602
InetSocketAddress address = InetSocketAddress.createUnresolved("localhost", 0);
@@ -798,10 +798,10 @@ public void testNodeIfOffline() {
798798

799799
TopicPartition tp = new TopicPartition("topic-1", 0);
800800

801-
assertOptional(metadata.fetch().nodeIfOnline(tp, 0), node -> assertEquals(node.id(), 0));
801+
assertOptional(metadata.fetch().nodeIfOnline(tp, 0), node -> assertEquals(0, node.id()));
802802
assertFalse(metadata.fetch().nodeIfOnline(tp, 1).isPresent());
803-
assertEquals(metadata.fetch().nodeById(0).id(), 0);
804-
assertEquals(metadata.fetch().nodeById(1).id(), 1);
803+
assertEquals(0, metadata.fetch().nodeById(0).id());
804+
assertEquals(1, metadata.fetch().nodeById(1).id());
805805
}
806806

807807
@Test
@@ -831,7 +831,7 @@ public void testNodeIfOnlineNonExistentTopicPartition() {
831831

832832
TopicPartition tp = new TopicPartition("topic-1", 0);
833833

834-
assertEquals(metadata.fetch().nodeById(0).id(), 0);
834+
assertEquals(0, metadata.fetch().nodeById(0).id());
835835
assertNull(metadata.fetch().partition(tp));
836836
assertEquals(metadata.fetch().nodeIfOnline(tp, 0), Optional.empty());
837837
}
@@ -955,13 +955,13 @@ protected boolean retainTopic(String topic, boolean isInternal, long nowMs) {
955955
// Update the metadata to add a new topic variant, "new", which will be retained with "keep". Note this
956956
// means that all of the "old" topics should be dropped.
957957
Cluster cluster = metadata.fetch();
958-
assertEquals(cluster.clusterResource().clusterId(), oldClusterId);
959-
assertEquals(cluster.nodes().size(), oldNodes);
958+
assertEquals(oldClusterId, cluster.clusterResource().clusterId());
959+
assertEquals(oldNodes, cluster.nodes().size());
960960
assertEquals(cluster.invalidTopics(), Set.of("oldInvalidTopic", "keepInvalidTopic"));
961961
assertEquals(cluster.unauthorizedTopics(), Set.of("oldUnauthorizedTopic", "keepUnauthorizedTopic"));
962962
assertEquals(cluster.topics(), Set.of("oldValidTopic", "keepValidTopic"));
963-
assertEquals(cluster.partitionsForTopic("oldValidTopic").size(), 2);
964-
assertEquals(cluster.partitionsForTopic("keepValidTopic").size(), 3);
963+
assertEquals(2, cluster.partitionsForTopic("oldValidTopic").size());
964+
assertEquals(3, cluster.partitionsForTopic("keepValidTopic").size());
965965
assertEquals(new HashSet<>(cluster.topicIds()), new HashSet<>(topicIds.values()));
966966

967967
String newClusterId = "newClusterId";
@@ -990,13 +990,13 @@ protected boolean retainTopic(String topic, boolean isInternal, long nowMs) {
990990
assertNull(metadataTopicIds2.get("oldValidTopic"));
991991

992992
cluster = metadata.fetch();
993-
assertEquals(cluster.clusterResource().clusterId(), newClusterId);
993+
assertEquals(newClusterId, cluster.clusterResource().clusterId());
994994
assertEquals(cluster.nodes().size(), newNodes);
995995
assertEquals(cluster.invalidTopics(), Set.of("keepInvalidTopic", "newInvalidTopic"));
996996
assertEquals(cluster.unauthorizedTopics(), Set.of("keepUnauthorizedTopic", "newUnauthorizedTopic"));
997997
assertEquals(cluster.topics(), Set.of("keepValidTopic", "newValidTopic"));
998-
assertEquals(cluster.partitionsForTopic("keepValidTopic").size(), 2);
999-
assertEquals(cluster.partitionsForTopic("newValidTopic").size(), 4);
998+
assertEquals(2, cluster.partitionsForTopic("keepValidTopic").size());
999+
assertEquals(4, cluster.partitionsForTopic("newValidTopic").size());
10001000
assertEquals(new HashSet<>(cluster.topicIds()), new HashSet<>(topicIds.values()));
10011001

10021002
// Perform another metadata update, but this time all topic metadata should be cleared.
@@ -1008,7 +1008,7 @@ protected boolean retainTopic(String topic, boolean isInternal, long nowMs) {
10081008
topicIds.forEach((topicName, topicId) -> assertNull(metadataTopicIds3.get(topicName)));
10091009

10101010
cluster = metadata.fetch();
1011-
assertEquals(cluster.clusterResource().clusterId(), newClusterId);
1011+
assertEquals(newClusterId, cluster.clusterResource().clusterId());
10121012
assertEquals(cluster.nodes().size(), newNodes);
10131013
assertEquals(cluster.invalidTopics(), Collections.emptySet());
10141014
assertEquals(cluster.unauthorizedTopics(), Collections.emptySet());

‎clients/src/test/java/org/apache/kafka/clients/admin/ConfigTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ public void shouldImplementEqualsProperly() {
6464
assertEquals(config, config);
6565
assertEquals(config, new Config(config.entries()));
6666
assertNotEquals(new Config(Collections.singletonList(E1)), config);
67-
assertNotEquals(config, "this");
67+
assertNotEquals("this", config);
6868
}
6969

7070
@Test

‎clients/src/test/java/org/apache/kafka/clients/consumer/CooperativeStickyAssignorTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ public void testEncodeAndDecodeGeneration() {
8181

8282
Optional<Integer> encodedGeneration = ((CooperativeStickyAssignor) assignor).memberData(subscription).generation;
8383
assertTrue(encodedGeneration.isPresent());
84-
assertEquals(encodedGeneration.get(), DEFAULT_GENERATION);
84+
assertEquals(DEFAULT_GENERATION, encodedGeneration.get());
8585

8686
int generation = 10;
8787
assignor.onAssignment(null, new ConsumerGroupMetadata("dummy-group-id", generation, "dummy-member-id", Optional.empty()));
@@ -90,7 +90,7 @@ public void testEncodeAndDecodeGeneration() {
9090
encodedGeneration = ((CooperativeStickyAssignor) assignor).memberData(subscription).generation;
9191

9292
assertTrue(encodedGeneration.isPresent());
93-
assertEquals(encodedGeneration.get(), generation);
93+
assertEquals(generation, encodedGeneration.get());
9494
}
9595

9696
@Test

‎clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -209,9 +209,7 @@ public void shouldReturnMaxPollRecords() {
209209
consumer.assign(Collections.singleton(partition));
210210
consumer.updateBeginningOffsets(Collections.singletonMap(partition, 0L));
211211

212-
IntStream.range(0, 10).forEach(offset -> {
213-
consumer.addRecord(new ConsumerRecord<>("test", 0, offset, null, null));
214-
});
212+
IntStream.range(0, 10).forEach(offset -> consumer.addRecord(new ConsumerRecord<>("test", 0, offset, null, null)));
215213

216214
consumer.setMaxPollRecords(2L);
217215

‎clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignorTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1025,7 +1025,7 @@ public void testAssignmentUpdatedForDeletedTopic(RackConfig rackConfig) {
10251025

10261026
Map<String, List<TopicPartition>> assignment = assignor.assignPartitions(partitionsPerTopic, subscriptions);
10271027
assertTrue(assignor.partitionsTransferringOwnership.isEmpty());
1028-
assertEquals(assignment.values().stream().mapToInt(List::size).sum(), 1 + 100);
1028+
assertEquals(1 + 100, assignment.values().stream().mapToInt(List::size).sum());
10291029
assertEquals(Collections.singleton(consumerId), assignment.keySet());
10301030
assertTrue(isFullyBalanced(assignment));
10311031
}
@@ -1043,7 +1043,7 @@ public void testNoExceptionThrownWhenOnlySubscribedTopicDeleted(RackConfig rackC
10431043

10441044
assignment = assignor.assign(Collections.emptyMap(), subscriptions);
10451045
assertTrue(assignor.partitionsTransferringOwnership.isEmpty());
1046-
assertEquals(assignment.size(), 1);
1046+
assertEquals(1, assignment.size());
10471047
assertTrue(assignment.get(consumerId).isEmpty());
10481048
}
10491049

0 commit comments

Comments
 (0)