chia7712 commented on code in PR #23990:
URL: https://github.com/apache/kafka/pull/23990#discussion_r4433795237
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DeleteTopicTest.java:
##########
@@ -267,7 +266,7 @@ public void testDeleteNonExistingTopic(ClusterInstance
cluster) throws Exception
cluster.waitTopicDeletion(topic);
waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas
for topic test not created.");
- AdminUtils.fetchOrWaitForLeader(admin, DEFAULT_TOPIC, 0, 1000);
+ AdminClientTestUtils.fetchOrWaitForLeader(admin, DEFAULT_TOPIC, 0,
1000);
Review Comment:
ditto. `waitForReplicaCreated` is a good guard.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DescribeProducersWithBrokerIdTest.java:
##########
@@ -78,7 +78,7 @@ private int getNonReplicaBrokerId(Admin admin) throws
Exception {
private int getFollowerBrokerId(Admin admin) throws Exception {
Review Comment:
Those methods are a bit wasteful to me since they keep calling the admin to
fetch the same information. We could just propagate TopicPartitionInfo since it
contains both the leader and replicas.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/AddPartitionsTest.java:
##########
@@ -108,8 +107,8 @@ public void testIncrementPartitions(ClusterInstance
cluster) throws Exception {
admin.createPartitions(Map.of("topic1",
NewPartitions.increaseTo(3))).all().get();
- AdminUtils.fetchOrWaitForLeader(admin, "topic1", 1, 30000);
- AdminUtils.fetchOrWaitForLeader(admin, "topic1", 2, 30000);
+ AdminClientTestUtils.fetchOrWaitForLeader(admin, "topic1", 1,
30000);
+ AdminClientTestUtils.fetchOrWaitForLeader(admin, "topic1", 2,
30000);
cluster.waitTopicCreation("topic1", 3);
Review Comment:
I think the wait for topic creation should be good enough for this test.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/ListOffsetsIntegrationTest.java:
##########
@@ -227,19 +227,16 @@ private void verifyListOffsets(String topic, int
expectedMaxTimestampOffset) thr
}
}, "There still are ongoing reassignments");
// make sure we are able to see the new leader
- TestUtils.waitForCondition(() -> {
- try {
- return clusterInstance.getLeaderBrokerId(new
TopicPartition(topic, 0)) == newLeader;
- } catch (InterruptedException | ExecutionException e) {
- return false;
- }
- }, () -> {
- try {
- return String.format("expected leader: %d but actual: %d",
newLeader, clusterInstance.getLeaderBrokerId(new TopicPartition(topic, 0)));
- } catch (ExecutionException | InterruptedException e) {
- throw new RuntimeException(e);
- }
- });
+ TestUtils.waitForCondition(
Review Comment:
the nested waiting is hard to read. Maybe we could add a small helper method
to fetch the leader instead
```java
private Optional<Integer> leader(TopicPartition tp) throws Exception {
return
adminClient.describeTopics(List.of(tp.topic())).allTopicNames().get()
.get(tp.topic()).partitions().stream()
.filter(p -> p.partition() == tp.partition())
.findFirst()
.map(TopicPartitionInfo::leader)
.filter(node -> node != null && node.id() != Node.noNode().id())
.map(Node::id);
}
```
BTW, I think `AdminClientTestUtils.fetchOrWaitForLeader` should be replaced
by this version. It just returns the current leader. Most tests don’t need to
wait for the leader to be ready, and the few that do can wrap it in
`waitForCondition`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
