diff --git a/pom.xml b/pom.xml
index a1c7cbd..2250813 100644
--- a/pom.xml
+++ b/pom.xml
@@ -74,7 +74,7 @@
         <dependency>
             <groupId>com.aliyun.openservices</groupId>
             <artifactId>aliyun-log</artifactId>
-            <version>0.6.75</version>
+            <version>0.6.87</version>
         </dependency>
     </dependencies>
 
diff --git a/src/main/java/com/aliyun/openservices/loghub/client/LogHubClientAdapter.java b/src/main/java/com/aliyun/openservices/loghub/client/LogHubClientAdapter.java
index 3a404a6..c3fe168 100755
--- a/src/main/java/com/aliyun/openservices/loghub/client/LogHubClientAdapter.java
+++ b/src/main/java/com/aliyun/openservices/loghub/client/LogHubClientAdapter.java
@@ -7,9 +7,11 @@
 import com.aliyun.openservices.log.common.ConsumerGroupShardCheckPoint;
 import com.aliyun.openservices.log.exception.LogException;
 import com.aliyun.openservices.log.http.client.ClientConfiguration;
+import com.aliyun.openservices.log.request.PullLogsRequest;
 import com.aliyun.openservices.log.response.BatchGetLogResponse;
 import com.aliyun.openservices.log.response.ConsumerGroupCheckPointResponse;
 import com.aliyun.openservices.log.response.ListConsumerGroupResponse;
+import com.aliyun.openservices.log.response.PullLogsResponse;
 import com.aliyun.openservices.loghub.client.config.LogHubConfig;
 import com.aliyun.openservices.loghub.client.config.LogHubCursorPosition;
 import com.aliyun.openservices.loghub.client.exceptions.LogHubClientWorkerException;
@@ -248,10 +250,17 @@ public String GetCursor(final int shard, final long time) throws LogException {
         }
     }
 
-    public BatchGetLogResponse BatchGetLogs(final int shard, final int lines, final String cursor) throws LogException {
+    public PullLogsResponse PullLogs(final int shard, final String cursor, LogHubConfig config) throws LogException {
         lock.readLock().lock();
         try {
-            return client.BatchGetLog(project, logstore, shard, lines, cursor);
+            int lines = config.getMaxFetchLogGroupSize();
+            PullLogsRequest pullLogsRequest = new PullLogsRequest(project, logstore, shard, lines, cursor);
+            String query = config.getQuery();
+            if (query != null && !query.isEmpty()) {
+                pullLogsRequest.setQuery(query);
+                pullLogsRequest.setPullMode("scan_on_stream");
+            }
+            return client.pullLogs(pullLogsRequest);
         } finally {
             lock.readLock().unlock();
         }
diff --git a/src/main/java/com/aliyun/openservices/loghub/client/LogHubFetchTask.java b/src/main/java/com/aliyun/openservices/loghub/client/LogHubFetchTask.java
index fdd1b2b..eec4862 100755
--- a/src/main/java/com/aliyun/openservices/loghub/client/LogHubFetchTask.java
+++ b/src/main/java/com/aliyun/openservices/loghub/client/LogHubFetchTask.java
@@ -2,9 +2,8 @@
 
 import com.aliyun.openservices.log.common.LogGroupData;
 import com.aliyun.openservices.log.exception.LogException;
-import com.aliyun.openservices.log.response.BatchGetLogResponse;
+import com.aliyun.openservices.log.response.PullLogsResponse;
 import com.aliyun.openservices.loghub.client.config.LogHubConfig;
-import com.aliyun.openservices.loghub.client.throttle.ResourceBarrier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -32,17 +31,16 @@ public TaskResult call() {
         Exception exception = null;
         for (int attempt = 0; ; attempt++) {
             try {
-                BatchGetLogResponse response = loghubClient.BatchGetLogs(
-                        shardId, config.getMaxFetchLogGroupSize(), cursor);
-                List<LogGroupData> fetchedData = response.GetLogGroups();
+                PullLogsResponse response = loghubClient.PullLogs(shardId, cursor, config);
+                List<LogGroupData> fetchedData = response.getLogGroups();
                 LOG.debug("shard {}, cursor {}, next cursor {}, response size: {}", shardId, cursor,
-                        response.GetNextCursor(), response.GetCount());
-                String nextCursor = response.GetNextCursor();
+                        response.getNextCursor(), response.getCount());
+                String nextCursor = response.getNextCursor();
                 if (nextCursor.isEmpty()) {
                     LOG.info("Shard {} next cursor is empty, set to current cursor {}", shardId, cursor);
                     nextCursor = cursor;
                 }
-                return new FetchTaskResult(fetchedData, cursor, nextCursor, response.GetRawSize());
+                return new FetchTaskResult(fetchedData, cursor, nextCursor, response.getRawSize());
             } catch (LogException lex) {
                 if (attempt == 0 && lex.GetErrorCode().toLowerCase().contains("invalidcursor")) {
                     // If checkpoint is invalid, such as expired cursor, will
diff --git a/src/main/java/com/aliyun/openservices/loghub/client/config/LogHubConfig.java b/src/main/java/com/aliyun/openservices/loghub/client/config/LogHubConfig.java
index 8f00376..c4c2007 100755
--- a/src/main/java/com/aliyun/openservices/loghub/client/config/LogHubConfig.java
+++ b/src/main/java/com/aliyun/openservices/loghub/client/config/LogHubConfig.java
@@ -23,6 +23,7 @@ public enum ConsumePosition {
     private String endpoint;
     private String project;
     private String logstore;
+    private String query;
     private String accessId;
     private String accessKey;
     private LogHubCursorPosition initialPosition;
@@ -74,6 +75,19 @@ public LogHubConfig(String consumerGroup,
         this.initialPosition = convertPosition(position);
     }
 
+    public LogHubConfig(String consumerGroup,
+                        String consumer,
+                        String endpoint,
+                        String project,
+                        String logstore,
+                        String accessId,
+                        String accessKey,
+                        ConsumePosition position,
+                        String query) {
+        this(consumerGroup, consumer, endpoint, project, logstore, accessId, accessKey, position);
+        this.setQuery(query);
+    }
+
     private static LogHubCursorPosition convertPosition(ConsumePosition position) {
         switch (position) {
             case BEGIN_CURSOR:
@@ -329,6 +343,14 @@ public void setMaxInProgressingDataSizeInMB(int maxInProgressingDataSizeInMB) {
         this.maxInProgressingDataSizeInMB = maxInProgressingDataSizeInMB;
     }
 
+    public String getQuery() {
+        return query;
+    }
+
+    public void setQuery(String query) {
+        this.query = query;
+    }
+
     @Override
     public String toString() {
         return "LogHubConfig{" +
@@ -337,6 +359,7 @@ public String toString() {
                 ", endpoint='" + endpoint + '\'' +
                 ", project='" + project + '\'' +
                 ", logstore='" + logstore + '\'' +
+                ", query ='" + query + '\'' +
                 ", accessId='" + accessId + '\'' +
                 ", initialPosition=" + initialPosition +
                 ", startTimestamp=" + startTimestamp +