hudi-agent commented on code in PR #20360:
URL: https://github.com/apache/hudi/pull/20360#discussion_r4290874025
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/KafkaTestUtils.java:
##########
@@ -89,13 +97,44 @@ public void createTopic(String topic, int numPartitions,
Properties properties)
).all().get();
}
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
Review Comment:
🤖 nit: `waitForPartitionLeaders` already declares `InterruptedException`, so
would it be cleaner to add a separate `catch (InterruptedException e)` than the
`instanceof` check inside the generic catch?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/testutils/KafkaTestUtils.java:
##########
@@ -89,13 +97,44 @@ public void createTopic(String topic, int numPartitions,
Properties properties)
).all().get();
}
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
if (e.getCause() instanceof TopicExistsException) {
throw (TopicExistsException) e.getCause();
}
throw new RuntimeException(e);
}
}
+ /**
+ * Topic creation returns before the broker has elected partition leaders.
Producing in that window
+ * gets NOT_LEADER_OR_FOLLOWER, and the idempotent producer's retries can
end in
+ * OUT_OF_ORDER_SEQUENCE_NUMBER, which drops records. Wait until every
partition has a leader.
+ */
+ private static void waitForPartitionLeaders(AdminClient adminClient, String
topic, int numPartitions)
+ throws ExecutionException, InterruptedException {
+ long deadline = System.nanoTime() + TOPIC_READY_TIMEOUT_NANOS;
+ while (true) {
+ try {
+ TopicDescription description =
+
adminClient.describeTopics(Collections.singleton(topic)).values().get(topic).get();
+ if (description.partitions().size() == numPartitions
+ && description.partitions().stream().allMatch(p -> p.leader() !=
null && !p.leader().isEmpty())) {
+ return;
+ }
+ } catch (ExecutionException e) {
+ if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) {
+ throw e;
+ }
+ }
+ if (System.nanoTime() - deadline > 0) {
+ throw new IllegalStateException("Timed out waiting for partition
leaders of topic " + topic);
Review Comment:
🤖 nit: could you pull the leader check into a small helper like
`hasAllPartitionLeaders(description, numPartitions)`? The condition inside the
try block is dense and nested.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
