gaborgsomogyi commented on code in PR #21247:
URL: https://github.com/apache/flink/pull/21247#discussion_r1017854901
##########
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java:
##########
@@ -88,8 +88,11 @@ public class KafkaSourceReaderTest extends
SourceReaderTestBase<KafkaPartitionSp
public static void setup() throws Throwable {
KafkaSourceTestEnv.setup();
try (AdminClient adminClient = KafkaSourceTestEnv.getAdminClient()) {
- adminClient.createTopics(
- Collections.singleton(new NewTopic(TOPIC, NUM_PARTITIONS,
(short) 1)));
+ adminClient
+ .createTopics(
+ Collections.singleton(new NewTopic(TOPIC,
NUM_PARTITIONS, (short) 1)))
+ .all()
+ .get();
// Use the admin client to trigger the creation of internal
__consumer_offsets topic.
// This makes sure that we won't see unavailable coordinator in
the tests.
waitUtil(
Review Comment:
In short yes. To give a little bit more detail:
* We create a general public topic w/ `createTopics` but the `waitUtil` is
waiting for an internal topic named `__consumer_offsets` to be created. They
have nothing to do w/ each other. Unless we consider the edge case where inside
Kafka there is no public topic and nobody enforced consumer group offset
storage.
* It's not written anywhere but according to my Kafka friends if one wants
stability then all places where the API gives back future one must wait for it.
* Pure personal opinion, I'm dealing w/ Kafka for ~5 years and if something
is forgotten to be synced then one can see such magical blow-ups which is
horror to catch :)
--
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]
