fix cpu load problem

apache · May 25, 2023 · c4a57ec · c4a57ec
1 parent 7d61ae9
commit c4a57ec
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 0 deletions.
diff --git a/...erver/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java b/...erver/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelSourceCollector.java
@@ -40,6 +40,8 @@ public class SeaTunnelSourceCollector<T> implements Collector<T> {
 
     private final Meter sourceReceivedQPS;
 
+    private volatile long rowCountThisPollNext;
+
     public SeaTunnelSourceCollector(
             Object checkpointLock,
             List<OneInputFlowLifeCycle<Record<?>>> outputs,
@@ -54,6 +56,7 @@ public SeaTunnelSourceCollector(
     public void collect(T row) {
         try {
             sendRecordToNext(new Record<>(row));
+            rowCountThisPollNext++;
             sourceReceivedCount.inc();
             sourceReceivedQPS.markEvent();
         } catch (IOException e) {
@@ -66,6 +69,16 @@ public Object getCheckpointLock() {
         return checkpointLock;
     }
 
+    @Override
+    public long getRowCountThisPollNext() {
+        return this.rowCountThisPollNext;
+    }
+
+    @Override
+    public void resetRowCountThisPollNext() {
+        this.rowCountThisPollNext = 0;
+    }
+
     public void sendRecordToNext(Record<?> record) throws IOException {
         synchronized (checkpointLock) {
             for (OneInputFlowLifeCycle<Record<?>> output : outputs) {

diff --git a/...erver/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java b/...erver/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java
@@ -133,6 +133,11 @@ public void close() throws IOException {
     public void collect() throws Exception {
         if (!prepareClose) {
             reader.pollNext(collector);
+            if (collector.getRowCountThisPollNext() == 0) {
+                Thread.sleep(100);
+            } else {
+                collector.resetRowCountThisPollNext();
+            }
         }
     }