Added code to throw exception if ODE continuation token goes into non…

… ODE pipeline

Microsoft.Azure.Cosmos/src/Query/Core/Pipeline/CosmosQueryExecutionContextFactory.cs

@@ -32,6 +32,7 @@ internal static class CosmosQueryExecutionContextFactory
     {
         private const string InternalPartitionKeyDefinitionProperty = "x-ms-query-partitionkey-definition";
         private const string OptimisticDirectExecution = "OptimisticDirectExecution";
+        private const string OptimisticDirectExecutionToken = "OptimisticDirectExecutionToken";
         private const string Passthrough = "Passthrough";
         private const string Specialized = "Specialized";
         private const int PageSizeFactorForTop = 5;
@@ -307,6 +308,19 @@ private static async Task<TryCatch<IQueryPipelineStage>> TryCreateFromPartitione
             }
             else
             {
+                if (inputParameters.InitialUserContinuationToken != null)
+                {
+                    CosmosObject cosmosObjectContinuationToken = inputParameters.InitialUserContinuationToken as CosmosObject;
+                    if (cosmosObjectContinuationToken.ContainsKey(OptimisticDirectExecutionToken))
+                    {
+                        return TryCatch<IQueryPipelineStage>.FromException(
+                            new MalformedContinuationTokenException(
+                                $"Operation cannot be resumed with the following token, as it requires the Optimistic Direct Execution pipeline which has been disabled on your SDK. " +
+                                $"Enable Optimistic Direct Execution from QueryRequestOptions to avoid this issue." +
+                                $"{inputParameters.InitialUserContinuationToken}."));
+                    }
+                }
+
                 if (createPassthroughQuery)
                 {
                     SetTestInjectionPipelineType(inputParameters, Passthrough);

Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/CosmosBasicQueryTests.cs

@@ -15,7 +15,6 @@ namespace Microsoft.Azure.Cosmos.SDK.EmulatorTests
     using Cosmos.Scripts;
     using Microsoft.Azure.Cosmos.Fluent;
     using Microsoft.Azure.Cosmos.Linq;
-    using Microsoft.Azure.Cosmos.Query.Core;
     using Microsoft.Azure.Cosmos.Tracing;
     using Microsoft.Azure.Documents.Collections;
     using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -786,6 +785,68 @@ public async Task QueryActivityIdWithContinuationTokenAndTraceTest()
 
         }
 
+        [TestMethod]
+        public async Task TesOdeTokenCompatibilityWithSpecializedPipeline()
+        {
+            using (ITrace rootTrace = Trace.GetRootTrace("Root Trace"))
+            {
+                ResponseMessage responseMessage = null;
+                int count = 0;
+                string query = "select top 200 * from c";
+                string expectedErrorMessage = "Operation cannot be resumed with the following token, as it requires Optimistic Direct Execution pipeline, which has been disabled for your SDK";
+
+                CosmosClient client = DirectCosmosClient;
+                Container container = client.GetContainer(DatabaseId, ContainerId);
+                // Create items
+                for (int i = 0; i < 500; i++)
+                {
+                    await container.CreateItemAsync<ToDoActivity>(ToDoActivity.CreateRandomToDoActivity());
+                }
+
+                QueryRequestOptions queryRequestOptions = new QueryRequestOptions
+                {
+                    MaxItemCount = 50,
+                    EnableOptimisticDirectExecution = true
+                };
+
+                FeedIteratorInternal feedIterator =
+                    (FeedIteratorInternal)container.GetItemQueryStreamIterator(
+                    query,
+                    null,
+                    queryRequestOptions);
+
+                responseMessage = await feedIterator.ReadNextAsync(rootTrace, CancellationToken.None);
+                string continuationToken = responseMessage.ContinuationToken;
+                Collection<ToDoActivity> response = TestCommon.SerializerCore.FromStream<CosmosFeedResponseUtil<ToDoActivity>>(responseMessage.Content).Data;
+                count += response.Count;
+
+                QueryRequestOptions newQueryRequestOptions = new QueryRequestOptions
+                {
+                    MaxItemCount = 50,
+                    EnableOptimisticDirectExecution = false
+                };
+
+                // use Continuation Token to create new iterator and use same trace
+                FeedIteratorInternal feedIteratorNew =
+                    (FeedIteratorInternal)container.GetItemQueryStreamIterator(
+                    query,
+                    continuationToken,
+                    newQueryRequestOptions);
+
+                while (feedIteratorNew.HasMoreResults)
+                {
+                    responseMessage = await feedIteratorNew.ReadNextAsync(rootTrace, CancellationToken.None);
+                    if (responseMessage.Content != null)
+                    {
+                        Collection<ToDoActivity> newResponse = TestCommon.SerializerCore.FromStream<CosmosFeedResponseUtil<ToDoActivity>>(responseMessage.Content).Data;
+                        count += newResponse.Count;
+                    }
+                }
+
+                Assert.IsTrue(responseMessage.CosmosException.ToString().Contains(expectedErrorMessage));
+            }
+        }
+
         private class CustomHandler : RequestHandler
         {
             string correlatedActivityId;