[hudi] branch master updated: [HUDI-4837] Stop sleeping where it is not necessary after the success (#6270)

[yihua]

This is an automated email from the ASF dual-hosted git repository.

yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new 35d03e9a1b [HUDI-4837] Stop sleeping where it is not necessary after 
the success (#6270)
35d03e9a1b is described below

commit 35d03e9a1bede05d10f10c6e4b57ffe66ca7f330
Author: Volodymyr Burenin <vbure...@gmail.com>
AuthorDate: Thu Sep 15 00:11:34 2022 -0500

    [HUDI-4837] Stop sleeping where it is not necessary after the success 
(#6270)
    
    Co-authored-by: Volodymyr Burenin <volodymyr.bure...@cloudkitchens.com>
    Co-authored-by: Y Ethan Guo <ethan.guoyi...@gmail.com>
---
 .../org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java     | 4 +++-
 .../java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java   | 4 ++--
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git 
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
 
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
index 1e78610ced..81f06a0f9f 100644
--- 
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
+++ 
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
@@ -315,7 +315,9 @@ public class KafkaOffsetGen {
       // TODO(HUDI-4625) cleanup, introduce retrying client
       partitionInfos = consumer.partitionsFor(topicName);
       try {
-        TimeUnit.SECONDS.sleep(10);
+        if (partitionInfos == null) {
+          TimeUnit.SECONDS.sleep(10);
+        }
       } catch (InterruptedException e) {
         LOG.error("Sleep failed while fetching partitions");
       }
diff --git 
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
 
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
index 2b99f19b27..1147736143 100644
--- 
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
+++ 
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
@@ -249,7 +249,7 @@ abstract class BaseTestKafkaSource extends 
SparkClientFunctionalTestHarness {
     // create a topic with very short retention
     final String topic = TEST_TOPIC_PREFIX + "testFailOnDataLoss";
     Properties topicConfig = new Properties();
-    topicConfig.setProperty("retention.ms", "10000");
+    topicConfig.setProperty("retention.ms", "8000");
     testUtils.createTopic(topic, 1, topicConfig);
 
     TypedProperties failOnDataLossProps = createPropsForKafkaSource(topic, 
null, "earliest");
@@ -261,7 +261,7 @@ abstract class BaseTestKafkaSource extends 
SparkClientFunctionalTestHarness {
     assertEquals(2, fetch1.getBatch().get().count());
 
     // wait for the checkpoint to expire
-    Thread.sleep(10001);
+    Thread.sleep(30000);
     Throwable t = assertThrows(HoodieDeltaStreamerException.class, () -> {
       
kafkaSource.fetchNewDataInAvroFormat(Option.of(fetch1.getCheckpointForNextBatch()),
 Long.MAX_VALUE);
     });