chia7712 commented on a change in pull request #9347:
URL: https://github.com/apache/kafka/pull/9347#discussion_r497017395
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java
##########
@@ -133,11 +133,14 @@ public void start() {
List<TopicPartition> partitions = new ArrayList<>();
// We expect that the topics will have been created either manually by
the user or automatically by the herder
- List<PartitionInfo> partitionInfos = null;
- long started = time.milliseconds();
- while (partitionInfos == null && time.milliseconds() - started <
CREATE_TOPIC_TIMEOUT_MS) {
+ List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
+ long started = time.nanoseconds();
+ long maxSleepMs = 1_000;
+ long sleepMs = 10;
+ while (partitionInfos == null && time.nanoseconds() - started <
CREATE_TOPIC_TIMEOUT_NS) {
+ time.sleep(sleepMs);
+ sleepMs = Math.min(2 * sleepMs, maxSleepMs);
partitionInfos = consumer.partitionsFor(topic);
Review comment:
It seems to me the behavior of ```timeout``` is not consistent in
consumer methods. The ```timeout``` used by other methods (for example:
```position```, ```offsetsForTimes```, ```beginningOffsets``` and
```endOffsets```) is to await the result of specify partitions. It means
consumer will send a request again if the timer is not expired and the specify
partition has no metadata (i.e topics data has not been propagated yet). Maybe
```partitionsFor``` should be fixed for consistent behavior.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
