diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index 9de676b..d9561fd 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -55,7 +55,6 @@
 public abstract class KylinConfigBase implements Serializable {
     private static final long serialVersionUID = 1L;
     private static final Logger logger = LoggerFactory.getLogger(KylinConfigBase.class);
-    private static final int AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();
 
     private static final String FALSE = "false";
     private static final String TRUE = "true";
@@ -2347,13 +2346,11 @@
     }
 
     public int getStreamingReceiverQueryCoreThreads() {
-        int def = getStreamingReceiverQueryMaxThreads() - 1;
-        return Integer.parseInt(getOptional("kylin.stream.receiver.query-core-threads", def + ""));
+        return Integer.parseInt(getOptional("kylin.stream.receiver.query-core-threads", "50"));
     }
 
     public int getStreamingReceiverQueryMaxThreads() {
-        int def = Math.max(2, AVAILABLE_PROCESSORS - 1);
-        return Integer.parseInt(getOptional("kylin.stream.receiver.query-max-threads", def + ""));
+        return Integer.parseInt(getOptional("kylin.stream.receiver.query-max-threads", "200"));
     }
 
     public int getStreamingReceiverUseThreadsPerQuery() {
diff --git a/storage-stream/src/main/java/org/apache/kylin/storage/stream/rpc/HttpStreamDataSearchClient.java b/storage-stream/src/main/java/org/apache/kylin/storage/stream/rpc/HttpStreamDataSearchClient.java
index e1119a8..bc075ce 100644
--- a/storage-stream/src/main/java/org/apache/kylin/storage/stream/rpc/HttpStreamDataSearchClient.java
+++ b/storage-stream/src/main/java/org/apache/kylin/storage/stream/rpc/HttpStreamDataSearchClient.java
@@ -72,11 +72,10 @@
  */
 public class HttpStreamDataSearchClient implements IStreamDataSearchClient {
     public static final Logger logger = LoggerFactory.getLogger(HttpStreamDataSearchClient.class);
-    public static final long WAIT_DURATION = 2 * 60000 ;
 
     private static ExecutorService executorService;
     static {
-        executorService = new ThreadPoolExecutor(20, 100, 60L, TimeUnit.SECONDS,
+        executorService = new ThreadPoolExecutor(20, 100, 60L, TimeUnit.MILLISECONDS,
                 new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("stream-rpc-pool-t"));
     }
     private AssignmentsCache assignmentsCache;
@@ -96,7 +95,7 @@
             final TupleFilter tupleFilter, final Set<TblColRef> dimensions, final Set<TblColRef> groups,
             final Set<FunctionDesc> metrics, final int storagePushDownLimit, final boolean allowStorageAggregation) {
         List<ReplicaSet> replicaSetsOfCube = assignmentsCache.getReplicaSetsByCube(cube.getName());
-        int timeout = cube.getConfig().getStreamingRPCHttpReadTimeout() * 2;
+        int timeout = 120 * 1000; // timeout should be configurable
         final QueuedStreamingTupleIterator result = new QueuedStreamingTupleIterator(replicaSetsOfCube.size(), timeout);
         final QueryContext query = QueryContextFacade.current();
 
@@ -106,6 +105,7 @@
         final RecordsSerializer recordsSerializer = new RecordsSerializer(schema);
         final DataRequest dataRequest = createDataRequest(query.getQueryId(), cube.getName(), minSegmentTime, tupleInfo,
                 tupleFilter, dimensions, groups, metrics, storagePushDownLimit, allowStorageAggregation);
+
         logger.info("Query-{}:send request to stream receivers", query.getQueryId());
         for (final ReplicaSet rs : replicaSetsOfCube) {
             executorService.submit(new Runnable() {
@@ -165,7 +165,7 @@
             return foundReceiver;
         }
 
-        if (System.currentTimeMillis() - lastFailTime > WAIT_DURATION) { // retry every 2 minutes
+        if (System.currentTimeMillis() - lastFailTime > 2 * 60 * 1000) { // retry every 2 minutes
             return foundReceiver;
         }
 
@@ -175,18 +175,18 @@
     public Iterator<ITuple> doSearch(DataRequest dataRequest, CubeInstance cube, StreamingTupleConverter tupleConverter,
             RecordsSerializer recordsSerializer, Node receiver, TupleInfo tupleInfo) throws Exception {
         String queryId = dataRequest.getQueryId();
+        logger.info("send query to receiver " + receiver + " with query id:" + queryId);
         String url = "http://" + receiver.getHost() + ":" + receiver.getPort() + "/kylin/api/data/query";
 
         try {
-            int connTimeout = cube.getConfig().getStreamingRPCHttpConnTimeout();
-            int readTimeout = cube.getConfig().getStreamingRPCHttpReadTimeout();
-            dataRequest.setDeadline(System.currentTimeMillis() + (int)(readTimeout * 1.5));
             String content = JsonUtil.writeValueAsString(dataRequest);
             Stopwatch sw = new Stopwatch();
             sw.start();
+            int connTimeout = cube.getConfig().getStreamingRPCHttpConnTimeout();
+            int readTimeout = cube.getConfig().getStreamingRPCHttpReadTimeout();
             String msg = restService.postRequest(url, content, connTimeout, readTimeout);
 
-            logger.info("query-{}: receive response from {} take time:{}", queryId, receiver, sw.elapsed(TimeUnit.MILLISECONDS));
+            logger.info("query-{}: receive response from {} take time:{}", queryId, receiver, sw.elapsedMillis());
             if (failedReceivers.containsKey(receiver)) {
                 failedReceivers.remove(receiver);
             }
diff --git a/stream-core/src/main/java/org/apache/kylin/stream/core/model/DataRequest.java b/stream-core/src/main/java/org/apache/kylin/stream/core/model/DataRequest.java
index f32b751..07c9028 100644
--- a/stream-core/src/main/java/org/apache/kylin/stream/core/model/DataRequest.java
+++ b/stream-core/src/main/java/org/apache/kylin/stream/core/model/DataRequest.java
@@ -36,7 +36,6 @@
     private boolean allowStorageAggregation;
 
     private long requestSendTime;
-    private long deadline;
     private boolean enableDetailProfile;
     private String storageBehavior;
 
@@ -143,12 +142,4 @@
     public void setHavingFilter(String havingFilter) {
         this.havingFilter = havingFilter;
     }
-
-    public long getDeadline() {
-        return deadline;
-    }
-
-    public void setDeadline(long deadline) {
-        this.deadline = deadline;
-    }
 }
diff --git a/stream-core/src/main/java/org/apache/kylin/stream/core/query/MultiThreadsResultCollector.java b/stream-core/src/main/java/org/apache/kylin/stream/core/query/MultiThreadsResultCollector.java
index 46f52a5..5ffe5f2 100644
--- a/stream-core/src/main/java/org/apache/kylin/stream/core/query/MultiThreadsResultCollector.java
+++ b/stream-core/src/main/java/org/apache/kylin/stream/core/query/MultiThreadsResultCollector.java
@@ -14,19 +14,18 @@
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
- */
+*/
 
 package org.apache.kylin.stream.core.query;
 
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Future;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.kylin.common.KylinConfig;
@@ -39,39 +38,28 @@
 
 public class MultiThreadsResultCollector extends ResultCollector {
     private static Logger logger = LoggerFactory.getLogger(MultiThreadsResultCollector.class);
-    private static ThreadPoolExecutor scannerThreadPool;
-    private static int MAX_RUNNING_THREAD_COUNT;
-
+    private static ExecutorService executor;
     static {
         KylinConfig config = KylinConfig.getInstanceFromEnv();
-        MAX_RUNNING_THREAD_COUNT = config.getStreamingReceiverQueryMaxThreads();
-        scannerThreadPool = new ThreadPoolExecutor(config.getStreamingReceiverQueryCoreThreads(),
-                MAX_RUNNING_THREAD_COUNT, 60L, TimeUnit.SECONDS,
-                new LinkedBlockingQueue<>(), new NamedThreadFactory("query-worker"));
+        executor = new ThreadPoolExecutor(config.getStreamingReceiverQueryCoreThreads(),
+                config.getStreamingReceiverQueryMaxThreads(), 60L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("query-worker"));
     }
 
-    private long deadline;
-    private String queryId;
-    /**
-     * if current query beyond the deadline
-     */
-    private AtomicBoolean cancelFlag = new AtomicBoolean(false);
+    private int timeout;
     private Semaphore workersSemaphore;
+    final BlockingQueue<Record> queue = new LinkedBlockingQueue<>(10000);
     private AtomicInteger notCompletedWorkers;
 
-    final BlockingQueue<Record> recordCachePool = new LinkedBlockingQueue<>(10000);
-
-    public MultiThreadsResultCollector(int numOfWorkers, long deadline) {
+    public MultiThreadsResultCollector(int numOfWorkers, int timeout) {
         this.workersSemaphore = new Semaphore(numOfWorkers);
-        this.deadline = deadline;
-        this.queryId = StreamingQueryProfile.get().getQueryId();
+        this.timeout = timeout;
     }
 
     @Override
     public Iterator<Record> iterator() {
         notCompletedWorkers = new AtomicInteger(searchResults.size());
-        Thread masterThread = new Thread(new WorkSubmitter(), "MultiThreadsResultCollector_" + queryId);
-        masterThread.start();
+        executor.submit(new WorkSubmitter());
 
         final int batchSize = 100;
         final long startTime = System.currentTimeMillis();
@@ -81,41 +69,38 @@
 
             @Override
             public boolean hasNext() {
-                boolean exits = (internalIT.hasNext() || !recordCachePool.isEmpty());
+                boolean exits = (internalIT.hasNext() || queue.size() > 0);
                 if (!exits) {
                     while (notCompletedWorkers.get() > 0) {
                         Thread.yield();
-                        if (System.currentTimeMillis() > deadline) {
-                            masterThread.interrupt(); // notify main thread
-                            cancelFlag.set(true);
-                            logger.warn("Beyond the deadline for {}.", queryId);
+                        long takeTime = System.currentTimeMillis() - startTime;
+                        if (takeTime > timeout) {
                             throw new RuntimeException("Timeout when iterate search result");
                         }
-                        if (internalIT.hasNext() || !recordCachePool.isEmpty()) {
+                        if (internalIT.hasNext() || queue.size() > 0) {
                             return true;
                         }
                     }
                 }
+
                 return exits;
             }
 
             @Override
             public Record next() {
                 try {
-                    if (System.currentTimeMillis() > deadline) {
+                    long takeTime = System.currentTimeMillis() - startTime;
+                    if (takeTime > timeout) {
                         throw new RuntimeException("Timeout when iterate search result");
                     }
                     if (!internalIT.hasNext()) {
                         recordList.clear();
-                        Record one = recordCachePool.poll(deadline - startTime, TimeUnit.MILLISECONDS);
+                        Record one = queue.poll(timeout - takeTime, TimeUnit.MILLISECONDS);
                         if (one == null) {
-                            masterThread.interrupt(); // notify main thread
-                            cancelFlag.set(true);
-                            logger.warn("Beyond the deadline for {}.", queryId);
                             throw new RuntimeException("Timeout when iterate search result");
                         }
                         recordList.add(one);
-                        recordCachePool.drainTo(recordList, batchSize - 1);
+                        queue.drainTo(recordList, batchSize - 1);
                         internalIT = recordList.iterator();
                     }
                     return internalIT.next();
@@ -143,13 +128,15 @@
             try {
                 result.startRead();
                 for (Record record : result) {
-                    recordCachePool.put(record.copy());
+                    try {
+                        queue.put(record.copy());
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException("Timeout when visiting streaming segmenent", e);
+                    }
                 }
                 result.endRead();
-            } catch (InterruptedException inter) {
-                logger.warn("Cancelled scan streaming segment", inter);
             } catch (Exception e) {
-                logger.error("Error when iterate search result", e);
+                logger.error("error when iterate search result", e);
             } finally {
                 notCompletedWorkers.decrementAndGet();
                 workersSemaphore.release();
@@ -160,44 +147,15 @@
     private class WorkSubmitter implements Runnable {
         @Override
         public void run() {
-            List<Future> futureList = Lists.newArrayListWithExpectedSize(searchResults.size());
-            int cancelTimes = 0;
-            try {
-                for (final IStreamingSearchResult result : searchResults) {
-                    Future f = scannerThreadPool.submit(new ResultIterateWorker(result));
-                    futureList.add(f);
-                    workersSemaphore.acquire(); // Throw InterruptedException when interrupted
-                }
-                while (notCompletedWorkers.get() > 0) {
-                    Thread.sleep(100);
-                    if (cancelFlag.get() || Thread.currentThread().isInterrupted()) {
-                        break;
-                    }
-                }
-            } catch (InterruptedException inter) {
-                logger.warn("Interrupted", inter);
-            } finally {
-                for (Future f : futureList) {
-                    if (!f.isCancelled() || !f.isDone()) {
-                        if (f.cancel(true)) {
-                            cancelTimes++;
-                        }
-                    }
+            for (final IStreamingSearchResult result : searchResults) {
+                executor.submit(new ResultIterateWorker(result));
+                try {
+                    workersSemaphore.acquire();
+                } catch (InterruptedException e) {
+                    logger.error("interrupted", e);
                 }
             }
-            logger.debug("Finish MultiThreadsResultCollector for queryId {}, cancel {}. Current thread pool: {}.",
-                    queryId, cancelTimes, scannerThreadPool);
         }
     }
 
-    /**
-     * block query if return true
-     */
-    public static boolean isOccupied() {
-        boolean occupied = scannerThreadPool.getActiveCount() >= MAX_RUNNING_THREAD_COUNT;
-        if (occupied) {
-            logger.debug("ThreadPool {}", scannerThreadPool);
-        }
-        return occupied;
-    }
 }
diff --git a/stream-core/src/main/java/org/apache/kylin/stream/core/query/StreamingCubeDataSearcher.java b/stream-core/src/main/java/org/apache/kylin/stream/core/query/StreamingCubeDataSearcher.java
index 1d38b90..f89ddec 100644
--- a/stream-core/src/main/java/org/apache/kylin/stream/core/query/StreamingCubeDataSearcher.java
+++ b/stream-core/src/main/java/org/apache/kylin/stream/core/query/StreamingCubeDataSearcher.java
@@ -43,6 +43,9 @@
 
 public class StreamingCubeDataSearcher {
     private static Logger logger = LoggerFactory.getLogger(StreamingCubeDataSearcher.class);
+
+    private static int TIMEOUT = Integer.MAX_VALUE;
+
     private StreamingSegmentManager streamingSegmentManager;
     private String cubeName;
     private CubeDesc cubeDesc;
@@ -70,15 +73,7 @@
         try {
             logger.info("query-{}: use cuboid {} to serve the query", queryProfile.getQueryId(),
                     searchRequest.getHitCuboid());
-            ResultCollector resultCollector = getResultCollector(searchRequest);
-            if (resultCollector instanceof MultiThreadsResultCollector) {
-                while (MultiThreadsResultCollector.isOccupied() && System.currentTimeMillis() < searchRequest.getDeadline()) {
-                    Thread.sleep(50);
-                }
-                if (System.currentTimeMillis() >= searchRequest.getDeadline()) {
-                    throw new RuntimeException("Timeout for " + queryProfile.getQueryId());
-                }
-            }
+            ResultCollector resultCollector = getResultCollector();
             Collection<StreamingCubeSegment> segments = streamingSegmentManager.getAllSegments();
             StreamingDataQueryPlanner scanRangePlanner = searchRequest.getQueryPlanner();
             for (StreamingCubeSegment queryableSegment : segments) {
@@ -110,10 +105,10 @@
         }
     }
 
-    private ResultCollector getResultCollector(StreamingSearchContext searchRequest) {
+    private ResultCollector getResultCollector() {
         int useThreads = cubeDesc.getConfig().getStreamingReceiverUseThreadsPerQuery();
         if (useThreads > 1) {
-            return new MultiThreadsResultCollector(useThreads, searchRequest.getDeadline());
+            return new MultiThreadsResultCollector(useThreads, TIMEOUT);
         } else {
             return new SingleThreadResultCollector();
         }
diff --git a/stream-core/src/main/java/org/apache/kylin/stream/core/query/StreamingSearchContext.java b/stream-core/src/main/java/org/apache/kylin/stream/core/query/StreamingSearchContext.java
index 307c2be..9da8e5b 100644
--- a/stream-core/src/main/java/org/apache/kylin/stream/core/query/StreamingSearchContext.java
+++ b/stream-core/src/main/java/org/apache/kylin/stream/core/query/StreamingSearchContext.java
@@ -45,8 +45,6 @@
     private long hitCuboid;
     private long basicCuboid;
 
-    private long deadline = Long.MAX_VALUE;
-
     public StreamingSearchContext(CubeDesc cubeDesc, Set<TblColRef> dimensions, Set<TblColRef> groups,
                                   Set<FunctionDesc> metrics, TupleFilter filter, TupleFilter havingFilter) {
         this.cubeDesc = cubeDesc;
@@ -58,7 +56,6 @@
         this.respResultSchema = new ResponseResultSchema(cubeDesc, dimensions, metrics);
         this.queryPlanner = new StreamingDataQueryPlanner(cubeDesc, filter);
         this.addedGroups = Sets.newHashSet();
-        this.deadline = deadline;
         calculateHitCuboid();
     }
 
@@ -161,12 +158,4 @@
         sortedSet.addAll(cubeDesc.getMandatoryCuboids());
         return sortedSet;
     }
-
-    public long getDeadline() {
-        return deadline;
-    }
-
-    public void setDeadline(long deadline) {
-        this.deadline =  deadline;
-    }
 }
diff --git a/stream-receiver/src/main/java/org/apache/kylin/stream/server/rest/controller/DataController.java b/stream-receiver/src/main/java/org/apache/kylin/stream/server/rest/controller/DataController.java
index 51d0cba..45c6307 100644
--- a/stream-receiver/src/main/java/org/apache/kylin/stream/server/rest/controller/DataController.java
+++ b/stream-receiver/src/main/java/org/apache/kylin/stream/server/rest/controller/DataController.java
@@ -24,7 +24,6 @@
 import org.apache.commons.codec.binary.Base64;
 import org.apache.kylin.common.KylinConfig;
 import org.apache.kylin.common.util.Pair;
-import org.apache.kylin.common.util.SetThreadName;
 import org.apache.kylin.cube.CubeDescManager;
 import org.apache.kylin.cube.model.CubeDesc;
 import org.apache.kylin.gridtable.StorageSideBehavior;
@@ -80,7 +79,7 @@
         }
         StreamingQueryProfile.set(queryProfile);
         logger.info("receive query request queryId:{}", queryId);
-        try (SetThreadName changeName = new SetThreadName("Query %s", queryId)) {
+        try {
             final Stopwatch sw = new Stopwatch();
             sw.start();
             String cubeName = dataRequest.getCubeName();
@@ -106,7 +105,6 @@
 
             StreamingSearchContext gtSearchRequest = new StreamingSearchContext(cubeDesc, dimensions, groups,
                     metrics, tupleFilter, havingFilter);
-            gtSearchRequest.setDeadline(dataRequest.getDeadline());
             searchResult = dataSearcher.doSearch(gtSearchRequest, minSegmentTime,
                     dataRequest.isAllowStorageAggregation());