diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/SourceStreamTask.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/SourceStreamTask.java
index 64fa332e0..c47fe63ae 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/SourceStreamTask.java
+++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/SourceStreamTask.java
@@ -30,9 +30,7 @@ public class SourceStreamTask extends StreamTask {
   public void run() {
     LOG.info("Source stream task thread start.");
 
-    while (running) {
-      sourceProcessor.run();
-    }
+    sourceProcessor.run();
   }
 
   @Override