Merge pull request #18 from taosdata/dev/zyyang

feat(subscribe): add timeout parameter#TD-33849

Author: zyyang90

src/main/java/com/taosdata/demo/service/Subscriber.java

@@ -32,6 +32,8 @@ public class Subscriber implements CommandLineRunner {
     private String consumerConfigFile;
     @Value("${subscriber.topic-names}")
     private String[] topicNames;
+    @Value("${subscriber.timeout}")
+    private String timeout;
     @Value("${subscriber.poll-timeout}")
     private int pollTimeout;
     @Value("${subscriber.print-data-in-log: false}")
@@ -102,10 +104,20 @@ private void consume(Properties properties) throws Exception {
             }
 
             int count = 0;
+            long start = System.currentTimeMillis();
+            long timeout_in_milli = Long.MAX_VALUE;
+            try {
+                timeout_in_milli = Long.parseLong(timeout);
+            } catch (Exception e) {
+                log.warn("failed to parse timeout: {}, use default: {}", timeout, timeout_in_milli);
+            }
             while (true) {
                 printOffsets(consumer, "before poll(" + count + ")");
                 try {
                     ConsumerRecords<Map<String, Object>> records = consumer.poll(Duration.ofMillis(pollTimeout));
+                    if (records.isEmpty() && System.currentTimeMillis() - start > timeout_in_milli) {
+                        break;
+                    }
                     for (ConsumerRecord<Map<String, Object>> record : records) {
                         String line = formatter.format(record);
                         if (printDataInLog) {

src/main/resources/application.yml

@@ -3,6 +3,8 @@ subscriber:
   concurrency: 1
   # 要订阅哪些topic，以逗号分隔
   topic-names: tp1,tp2,tp3
+  # 订阅超时，超过这个时间没有订阅到数据就会退出
+  timeout: 5000
   # 每次 pull 的超时时间
   poll-timeout: 1000
   # pull 以后，写到一个文件里