Savonitar commented on code in PR #237:
URL:
https://github.com/apache/flink-connector-kafka/pull/237#discussion_r2955680643
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java:
##########
@@ -144,38 +142,7 @@ public void createTestTopic(
public static void createNewTopic(
String topic, int numberOfPartitions, int replicationFactor,
Properties properties) {
- LOG.info("Creating topic {}", topic);
- try (AdminClient adminClient = AdminClient.create(properties)) {
- NewTopic topicObj = new NewTopic(topic, numberOfPartitions,
(short) replicationFactor);
-
adminClient.createTopics(Collections.singleton(topicObj)).all().get();
- CommonTestUtils.waitUtil(
- () -> {
- try {
- // Ensure all partitions have a leader elected and
logs initialized
- Map<TopicPartition, OffsetSpec> offsetSpecs = new
HashMap<>();
- for (int i = 0; i < numberOfPartitions; i++) {
- offsetSpecs.put(
- new TopicPartition(topic, i),
OffsetSpec.earliest());
- }
- adminClient
- .listOffsets(offsetSpecs)
- .all()
- .get(REQUEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
- return true;
- } catch (Exception e) {
- LOG.warn(
- "Partitions for topic {} not yet ready to
serve requests",
- topic,
- e);
- return false;
- }
- },
- Duration.ofSeconds(30),
- String.format("New topic \"%s\" is not ready within
timeout", topicObj));
- } catch (Exception e) {
- e.printStackTrace();
- fail("Create test topic : " + topic + " failed, " +
e.getMessage());
- }
+ createNewTopicAndWaitForPartitionAssignment(topic, numberOfPartitions,
replicationFactor, properties);
Review Comment:
What is the motivation to keep it not inlined?
--
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]
