Cosmos SDK v4 - Pagination based on continuation token

[rkganji]

We have a requirement for clients to get results in paginated way. Pagination was implemented as below:

final CosmosQueryRequestOptions options = new CosmosQueryRequestOptions();
options.setResponseContinuationTokenLimitInKb(2);
options.setQueryMetricsEnabled(true);
        
// Retrieve the documents response using DocumentClient.
Iterable<FeedResponse<Reservation>> feedResponseIterator = container
                .queryItems(query, options, Reservation.class)
                .iterableByPage(continuationToken, 50);
List<Object> docsList = new ArrayList<>();
ResponseWithHeaders result = new ResponseWithHeaders();
FeedResponse<Reservation> page = feedResponseIterator.iterator().next();
String nextToken = page.getContinuationToken();
docsList.addAll(page.getResults());

Issue:
This is working fine when overall result set is less e.g., 120 documents returned as 3 pages. This is not working fine when overall results set is large. When total no. of documents is ~1000, same continuationToken is being returned causing an endless loop and not able to get documents beyond initial 50 docs returned.

[ghost]

Thanks for the feedback! We are routing this to the appropriate team for follow-up. cc @wmengmsft, @MehaKaushik, @shurd, @anfeldma-ms

[joshfree]

Thanks for filing this Cosmos v4 sdk issue, @rkganji. Someone from the Cosmos team will follow up shortly.

/cc @kushagraThapar

[kushagraThapar]

@rkganji - can you please share the query that you are running ?

[rkganji]

@kushagraThapar Query based on our project schema look like below (Basically, search reservations in a certain timeframe)

SELECT value r FROM ROOT r JOIN rs IN r.roomStays.roomStay WHERE rs.timeSpan.startTime>=1597017600000 AND rs.timeSpan.startTime<1597104000000 AND r.hotelCode="001"

[kushagraThapar]

@rkganji - how are you using the next continuation token ?
Also, did you try iterating over the results without the continuation token and just iterating over the result set directly.

[rkganji]

@kushagraThapar Our API returns the continuation token to our internal client along with results. Clients iteratively calls our API by passing continuation token returned by our API until there's no continuation token available. Basically, clients iteratively call our API to get all the results in a paginated way.

If we do below, things work fine as we are getting all results in one shot and there's no pagination as such:

final CosmosQueryRequestOptions options = new CosmosQueryRequestOptions();
options.setResponseContinuationTokenLimitInKb(Integer.parseInt(System.getenv("continuationTokenLimitInKb")));
options.setQueryMetricsEnabled(true);

// Retrieve the documents response using DocumentClient.
Iterable<FeedResponse<Reservation>> feedResponseIterator = container
                .queryItems(query, options, Reservation.class)
                .iterableByPage(continuationToken, Integer.parseInt(System.getenv("maxItemCount")));

List<Object> docsList = new ArrayList<>();
ResponseWithHeaders result = new ResponseWithHeaders();
String nextToken = null;
for (FeedResponse<Reservation> page : feedResponseIterator) {
      docsList.addAll(page.getResults());
}

[kushagraThapar]

I see, makes sense. What version of SDK are you using ?

[rkganji]

We are using 4.3.0

[kushagraThapar]

@rkganji - We have tried to fix this issue here - #16775
Once the new release goes out, you should be able to test it against the newly released version, will keep you updated.

[rkganji]

@kushagraThapar Great, Thanks for the update

[kushagraThapar]

@rkganji - can you please confirm if this fixes your issue ?

[rkganji]

@kushagraThapar On trying to integrate with v4.8.0, I get below error when initialising CosmosClient. I don't get this issue on 4.7.1 or 4.7.0. Something underlying changed?

18:54:36.225 [parallel-1] ERROR reactor.core.scheduler.Schedulers - Scheduler worker in group main failed with an uncaught exception [11/03/2020 02:54:36] java.lang.NoSuchMethodError: io.netty.channel.SingleThreadEventLoop.<init>(Lio/netty/channel/EventLoopGroup;Ljava/util/concurrent/Executor;ZLjava/util/Queue;Ljava/util/Queue;Lio/netty/util/concurrent/RejectedExecutionHandler;)V [11/03/2020 02:54:36] at io.netty.channel.kqueue.KQueueEventLoop.<init>(KQueueEventLoop.java:77) ~[netty-transport-native-kqueue-4.1.51.Final-osx-x86_64.jar:4.1.51.Final] [11/03/2020 02:54:36] at io.netty.channel.kqueue.KQueueEventLoopGroup.newChild(KQueueEventLoopGroup.java:151) ~[netty-transport-native-kqueue-4.1.51.Final-osx-x86_64.jar:4.1.51.Final] [11/03/2020 02:54:36] at io.netty.channel.kqueue.KQueueEventLoopGroup.newChild(KQueueEventLoopGroup.java:32) ~[netty-transport-native-kqueue-4.1.51.Final-osx-x86_64.jar:4.1.51.Final] [11/03/2020 02:54:36] at io.netty.util.concurrent.MultithreadEventExecutorGroup.<init>(MultithreadEventExecutorGroup.java:84) ~[azure-functions-java-worker.jar:1.5.2-SNAPSHOT] [11/03/2020 02:54:36] at io.netty.util.concurrent.MultithreadEventExecutorGroup.<init>(MultithreadEventExecutorGroup.java:58) ~[azure-functions-java-worker.jar:1.5.2-SNAPSHOT] [11/03/2020 02:54:36] at io.netty.util.concurrent.MultithreadEventExecutorGroup.<init>(MultithreadEventExecutorGroup.java:47) ~[azure-functions-java-worker.jar:1.5.2-SNAPSHOT] [11/03/2020 02:54:36] at io.netty.channel.MultithreadEventLoopGroup.<init>(MultithreadEventLoopGroup.java:59) ~[azure-functions-java-worker.jar:1.5.2-SNAPSHOT] [11/03/2020 02:54:36] at io.netty.channel.kqueue.KQueueEventLoopGroup.<init>(KQueueEventLoopGroup.java:110) ~[netty-transport-native-kqueue-4.1.51.Final-osx-x86_64.jar:4.1.51.Final] [11/03/2020 02:54:36] at io.netty.channel.kqueue.KQueueEventLoopGroup.<init>(KQueueEventLoopGroup.java:97) ~[netty-transport-native-kqueue-4.1.51.Final-osx-x86_64.jar:4.1.51.Final] [11/03/2020 02:54:36] at io.netty.channel.kqueue.KQueueEventLoopGroup.<init>(KQueueEventLoopGroup.java:73) ~[netty-transport-native-kqueue-4.1.51.Final-osx-x86_64.jar:4.1.51.Final] [11/03/2020 02:54:36] at reactor.netty.resources.DefaultLoopKQueue.newEventLoopGroup(DefaultLoopKQueue.java:77) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.resources.DefaultLoopResources.cacheNativeServerLoops(DefaultLoopResources.java:255) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.resources.DefaultLoopResources.cacheNativeClientLoops(DefaultLoopResources.java:270) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.resources.DefaultLoopResources.onClient(DefaultLoopResources.java:202) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpResources.onClient(TcpResources.java:234) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.http.client.HttpClientConnect$HttpTcpClient.connect(HttpClientConnect.java:139) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClientOperator.connect(TcpClientOperator.java:43) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.tcp.TcpClient.connect(TcpClient.java:218) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.http.client.HttpClientFinalizer.connect(HttpClientFinalizer.java:80) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at reactor.netty.http.client.HttpClientFinalizer.responseConnection(HttpClientFinalizer.java:97) ~[reactor-netty-0.9.11.RELEASE.jar:0.9.11.RELEASE] [11/03/2020 02:54:36] at com.azure.cosmos.implementation.http.ReactorNettyClient.send(ReactorNettyClient.java:135) ~[azure-cosmos-4.8.0.jar:?] [11/03/2020 02:54:36] at com.azure.cosmos.implementation.RxGatewayStoreModel.performRequest(RxGatewayStoreModel.java:168) ~[azure-cosmos-4.8.0.jar:?] [11/03/2020 02:54:36] at com.azure.cosmos.implementation.RxGatewayStoreModel.read(RxGatewayStoreModel.java:95) ~[azure-cosmos-4.8.0.jar:?] [11/03/2020 02:54:36] at com.azure.cosmos.implementation.RxGatewayStoreModel.invokeAsyncInternal(RxGatewayStoreModel.java:363) ~[azure-cosmos-4.8.0.jar:?] [11/03/2020 02:54:36] at com.azure.cosmos.implementation.RxGatewayStoreModel.lambda$invokeAsync$4(RxGatewayStoreModel.java:378) ~[azure-cosmos-4.8.0.jar:?] [11/03/2020 02:54:36] at com.azure.cosmos.implementation.BackoffRetryUtility.lambda$executeRetry$0(BackoffRetryUtility.java:36) ~[azure-cosmos-4.8.0.jar:?] [11/03/2020 02:54:36] at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:44) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.FluxRetryWhen.subscribe(FluxRetryWhen.java:79) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.MonoRetryWhen.subscribeOrReturn(MonoRetryWhen.java:46) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.Mono.subscribe(Mono.java:4198) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.FluxFlatMap.trySubscribeScalarMap(FluxFlatMap.java:199) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.MonoFlatMap.subscribeOrReturn(MonoFlatMap.java:53) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.FluxFromMonoOperator.subscribe(FluxFromMonoOperator.java:76) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.FluxDefer.subscribe(FluxDefer.java:54) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:117) ~[reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68) [reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28) [reactor-core-3.3.9.RELEASE.jar:3.3.9.RELEASE] [11/03/2020 02:54:36] at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_201] [11/03/2020 02:54:36] at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_201] [11/03/2020 02:54:36] at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [?:1.8.0_201] [11/03/2020 02:54:36] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_201] [11/03/2020 02:54:36] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_201] [11/03/2020 02:54:36] at java.lang.Thread.run(Thread.java:748) [?:1.8.0_201] [11/03/2020 02:56:35] 18:56:35.931 [pool-2-thread-4] ERROR com.azure.cosmos.implementation.RxDocumentClientImpl - unexpected failure in initializing client.

[kushagraThapar]

@rkganji - thanks for the stack trace, looks like a netty dependency issue. I will investigate this.

[kushagraThapar]

@rkganji - I verified the netty versions issues from Cosmos SDK side, and I don't see any issues there.
I also updated the samples and getting started guide with new version of azure-cosmos. You can verify the version by running these samples : https://github.com/Azure-Samples/azure-cosmos-java-getting-started
and https://github.com/Azure-Samples/azure-cosmos-java-sql-api-samples

[rkganji]

@kushagraThapar May be some version mismatch running from functions runtime. I will investigate this.

[rkganji]

@kushagraThapar Netty issue that I reported is due to Azure/azure-functions-java-worker#381 which I was able to resolve by upgrading azure functions core tools and using FUNCTIONS_WORKER_JAVA_LOAD_APP_LIBS as mentioned to override class loading behaviour for azure functions.

I tested the pagination fix and found to be working fine now.

[kushagraThapar]

@rkganji - thanks for the verification, closing this issue now.