Adding basic FeedRanges API

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java

@@ -3682,7 +3682,7 @@ private static List<FeedRange> toFeedRanges(
             throw new IllegalStateException("PartitionKeyRange list cannot be null");
         }
 
-        List<FeedRange> feedRanges = new ArrayList<FeedRange>();
+        List<FeedRange> feedRanges = new ArrayList<FeedRange>(partitionKeyRangeList.size());
         partitionKeyRangeList.forEach(pkRange -> {
             feedRanges.add(toFeedRange(pkRange));
         });

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ShouldRetryResult.java

@@ -1,3 +1,5 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
 package com.azure.cosmos.implementation;
 
 import java.time.Duration;

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/feedranges/FeedRangeAsyncVisitor.java

@@ -6,9 +6,9 @@
 import reactor.core.publisher.Mono;
 
 abstract class FeedRangeAsyncVisitor<TResult> {
-    public abstract Mono<TResult> visitAsync(FeedRangePartitionKeyImpl feedRange);
+    public abstract Mono<TResult> visit(FeedRangePartitionKeyImpl feedRange);
 
-    public abstract Mono<TResult> visitAsync(FeedRangePartitionKeyRangeImpl feedRange);
+    public abstract Mono<TResult> visit(FeedRangePartitionKeyRangeImpl feedRange);
 
-    public abstract Mono<TResult> visitAsync(FeedRangeEpkImpl feedRange);
+    public abstract Mono<TResult> visit(FeedRangeEpkImpl feedRange);
 }

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/feedranges/FeedRangeAsyncVisitorWithArg.java

@@ -6,10 +6,10 @@
 import reactor.core.publisher.Mono;
 
 abstract class FeedRangeAsyncVisitorWithArg<TResult, TArg> {
-    public abstract Mono<TResult> visitAsync(FeedRangePartitionKeyImpl feedRange, TArg argument);
+    public abstract Mono<TResult> visit(FeedRangePartitionKeyImpl feedRange, TArg argument);
 
-    public abstract Mono<TResult> visitAsync(FeedRangePartitionKeyRangeImpl feedRange,
-                                             TArg argument);
+    public abstract Mono<TResult> visit(FeedRangePartitionKeyRangeImpl feedRange,
+                                        TArg argument);
 
-    public abstract Mono<TResult> visitAsync(FeedRangeEpkImpl feedRange, TArg argument);
+    public abstract Mono<TResult> visit(FeedRangeEpkImpl feedRange, TArg argument);
 }

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/feedranges/FeedRangeCompositeContinuationImpl.java

@@ -12,9 +12,12 @@
 import com.azure.cosmos.implementation.Strings;
 import com.azure.cosmos.implementation.Utils;
 import com.azure.cosmos.implementation.caches.RxPartitionKeyRangeCache;
+import com.azure.cosmos.implementation.directconnectivity.GatewayAddressCache;
 import com.azure.cosmos.implementation.query.CompositeContinuationToken;
 import com.azure.cosmos.implementation.routing.Range;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
 
 import java.io.IOException;
@@ -32,6 +35,7 @@
  */
 final class FeedRangeCompositeContinuationImpl extends FeedRangeContinuation {
 
+    private final static Logger LOGGER = LoggerFactory.getLogger(FeedRangeCompositeContinuationImpl.class);
     private final static ShouldRetryResult NO_RETRY = ShouldRetryResult.noRetry();
     private final static ShouldRetryResult RETRY = ShouldRetryResult.retryAfter(Duration.ZERO);
     private final Queue<CompositeContinuationToken> compositeContinuationTokens;
@@ -139,10 +143,7 @@ public void validateContainer(final String containerRid) throws IllegalArgumentE
 
     @Override
     public ShouldRetryResult handleChangeFeedNotModified(final RxDocumentServiceResponse response) {
-        if (response == null) {
-            throw new NullPointerException("response");
-        }
-
+        checkNotNull(response, "Argument 'response' must not be null");
         final int statusCode = response.getStatusCode();
         if (statusCode >= HttpConstants.StatusCodes.MINIMUM_SUCCESS_STATUSCODE
             && statusCode <= HttpConstants.StatusCodes.MAXIMUM_SUCCESS_STATUSCODE) {
@@ -171,16 +172,11 @@ public ShouldRetryResult handleChangeFeedNotModified(final RxDocumentServiceResp
     }
 
     @Override
-    public Mono<ShouldRetryResult> handleSplitAsync(final RxDocumentClientImpl client,
-                                                    final RxDocumentServiceResponse response) {
+    public Mono<ShouldRetryResult> handleSplit(final RxDocumentClientImpl client,
+                                               final RxDocumentServiceResponse response) {
 
-        if (client == null) {
-            throw new NullPointerException("client");
-        }
-
-        if (response == null) {
-            throw new NullPointerException("response");
-        }
+        checkNotNull(client, "Argument 'client' must not be null");
+        checkNotNull(response, "Argument 'response' must not be null");
 
         Integer nSubStatus = 0;
         final String valueSubStatus =
@@ -200,7 +196,7 @@ public Mono<ShouldRetryResult> handleSplitAsync(final RxDocumentClientImpl clien
 
         final RxPartitionKeyRangeCache partitionKeyRangeCache = client.getPartitionKeyRangeCache();
         final Mono<Utils.ValueHolder<List<PartitionKeyRange>>> resolvedRangesTask =
-            this.tryGetOverlappingRangesAsync(
+            this.tryGetOverlappingRanges(
                 partitionKeyRangeCache, this.currentToken.getRange().getMin(),
                 this.currentToken.getRange().getMax(),
                 true);
@@ -216,10 +212,7 @@ public Mono<ShouldRetryResult> handleSplitAsync(final RxDocumentClientImpl clien
 
     @Override
     public void accept(final FeedRangeContinuationVisitor visitor) {
-        if (visitor == null) {
-            throw new NullPointerException("visitor");
-        }
-
+        checkNotNull(visitor, "Argument 'visitor' must not be null");
         visitor.visit(this);
     }
 
@@ -248,12 +241,8 @@ public static FeedRangeCompositeContinuationImpl createFromDeserializedTokens(
     }
 
     public static FeedRangeContinuation parse(final String jsonString) throws IOException {
-        if (jsonString == null) {
-            throw new NullPointerException("jsonString");
-        }
-
+        checkNotNull(jsonString, "Argument 'jsonString' must not be null");
         final ObjectMapper mapper = Utils.getSimpleObjectMapper();
-
         return mapper.readValue(jsonString, FeedRangeCompositeContinuationImpl.class);
     }
 
@@ -330,7 +319,7 @@ private void moveToNextToken() {
         }
     }
 
-    private Mono<Utils.ValueHolder<List<PartitionKeyRange>>> tryGetOverlappingRangesAsync(
+    private Mono<Utils.ValueHolder<List<PartitionKeyRange>>> tryGetOverlappingRanges(
         final RxPartitionKeyRangeCache partitionKeyRangeCache, final String min, final String max,
         final Boolean forceRefresh) {
 
@@ -360,6 +349,10 @@ private static CompositeContinuationToken tryParseAsCompositeContinuationToken(
 
             return null;
         } catch (final IOException ioError) {
+            LOGGER.debug(
+                "Failed to parse as composite continuation token JSON ",
+                providedContinuation,
+                ioError);
             return null;
         }
     }

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/feedranges/FeedRangeContinuation.java

@@ -50,7 +50,7 @@ public static FeedRangeContinuation tryParse(String toStringValue) {
     public abstract ShouldRetryResult handleChangeFeedNotModified(
         RxDocumentServiceResponse responseMessage);
 
-    public abstract Mono<ShouldRetryResult> handleSplitAsync(
+    public abstract Mono<ShouldRetryResult> handleSplit(
         RxDocumentClientImpl client,
         RxDocumentServiceResponse responseMessage);
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/feedranges/FeedRangeEpkImpl.java

@@ -18,6 +18,7 @@
 import java.util.Objects;
 
 import static com.azure.cosmos.BridgeInternal.setProperty;
+import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkNotNull;
 
 final class FeedRangeEpkImpl extends FeedRangeInternal {
     private static final FeedRangeEpkImpl fullRangeEPK =
@@ -27,10 +28,7 @@ final class FeedRangeEpkImpl extends FeedRangeInternal {
     private final UnmodifiableList<Range<String>> rangeList;
 
     public FeedRangeEpkImpl(final Range<String> range) {
-        if (range == null) {
-            throw new NullPointerException("range");
-        }
-
+        checkNotNull(range, "Argument 'range' must not be null");
         this.range = range;
         final ArrayList<Range<String>> temp = new ArrayList<>();
         temp.add(range);
@@ -48,33 +46,24 @@ public static FeedRangeEpkImpl forFullRange() {
 
     @Override
     public void accept(final FeedRangeVisitor visitor) {
-        if (visitor == null) {
-            throw new NullPointerException("visitor");
-        }
-
+        checkNotNull(visitor, "Argument 'visitor' must not be null");
         visitor.visit(this);
     }
 
     @Override
     public <TInput> void accept(GenericFeedRangeVisitor<TInput> visitor, TInput input) {
-        if (visitor == null) {
-            throw new NullPointerException("visitor");
-        }
-
+        checkNotNull(visitor, "Argument 'visitor' must not be null");
         visitor.visit(this, input);
     }
 
     @Override
-    public <T> Mono<T> acceptAsync(final FeedRangeAsyncVisitor<T> visitor) {
-        if (visitor == null) {
-            throw new NullPointerException("visitor");
-        }
-
-        return visitor.visitAsync(this);
+    public <T> Mono<T> accept(final FeedRangeAsyncVisitor<T> visitor) {
+        checkNotNull(visitor, "Argument 'visitor' must not be null");
+        return visitor.visit(this);
     }
 
     @Override
-    public Mono<UnmodifiableList<Range<String>>> getEffectiveRangesAsync(
+    public Mono<UnmodifiableList<Range<String>>> getEffectiveRanges(
         final IRoutingMapProvider routingMapProvider,
         final String containerRid,
         final PartitionKeyDefinition partitionKeyDefinition) {
@@ -83,7 +72,7 @@ public Mono<UnmodifiableList<Range<String>>> getEffectiveRangesAsync(
     }
 
     @Override
-    public Mono<UnmodifiableList<String>> getPartitionKeyRangesAsync(
+    public Mono<UnmodifiableList<String>> getPartitionKeyRanges(
         final IRoutingMapProvider routingMapProvider,
         final String containerRid,
         final PartitionKeyDefinition partitionKeyDefinition) {

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/feedranges/FeedRangeInternal.java

@@ -15,23 +15,25 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
 
 import java.io.IOException;
 
+import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkNotNull;
+
 public abstract class FeedRangeInternal extends JsonSerializable implements FeedRange {
+    private final static Logger LOGGER = LoggerFactory.getLogger(FeedRangeInternal.class);
 
     public abstract void accept(FeedRangeVisitor visitor);
 
     public abstract <TInput> void accept(GenericFeedRangeVisitor<TInput> visitor, TInput input);
 
-    public abstract <T> Mono<T> acceptAsync(FeedRangeAsyncVisitor<T> visitor);
+    public abstract <T> Mono<T> accept(FeedRangeAsyncVisitor<T> visitor);
 
     public static FeedRangeInternal convert(final FeedRange feedRange) {
-        if (feedRange == null) {
-            throw new NullPointerException("feedRange");
-        }
-
+        checkNotNull(feedRange, "Argument 'feedRange' must not be null");
         if (feedRange instanceof FeedRangeInternal) {
             return (FeedRangeInternal)feedRange;
         }
@@ -47,14 +49,7 @@ public static FeedRangeInternal convert(final FeedRange feedRange) {
      * @return A feed range
      */
     public static FeedRangeInternal fromJsonString(String json) {
-        FeedRangeInternal parsedRange;
-
-        try {
-            parsedRange = FeedRangeInternal.tryParse(json);
-        } catch (IOException e) {
-            throw new IllegalArgumentException(
-                String.format("Unable to parse JSON %s", json), e);
-        }
+        FeedRangeInternal parsedRange = FeedRangeInternal.tryParse(json);
 
         if (parsedRange == null) {
             throw new IllegalArgumentException(
@@ -66,12 +61,12 @@ public static FeedRangeInternal fromJsonString(String json) {
         return parsedRange;
     }
 
-    public abstract Mono<UnmodifiableList<Range<String>>> getEffectiveRangesAsync(
+    public abstract Mono<UnmodifiableList<Range<String>>> getEffectiveRanges(
         IRoutingMapProvider routingMapProvider,
         String containerRid,
         PartitionKeyDefinition partitionKeyDefinition);
 
-    public abstract Mono<UnmodifiableList<String>> getPartitionKeyRangesAsync(
+    public abstract Mono<UnmodifiableList<String>> getPartitionKeyRanges(
         IRoutingMapProvider routingMapProvider,
         String containerRid,
         PartitionKeyDefinition partitionKeyDefinition);
@@ -88,32 +83,36 @@ public String toJsonString() {
         return this.toJson();
     }
 
-    public static FeedRangeInternal tryParse(final String jsonString) throws IOException {
-        if (jsonString == null) {
-            throw new NullPointerException("jsonString");
-        }
-
+    public static FeedRangeInternal tryParse(final String jsonString) {
+        checkNotNull(jsonString, "Argument 'jsonString' must not be null");
         final ObjectMapper mapper = Utils.getSimpleObjectMapper();
-        JsonNode rootNode = mapper.readTree(jsonString);
-
-        JsonNode rangeNode = rootNode.get(Constants.Properties.RANGE);
-        if (rangeNode != null && rangeNode.isObject()) {
-            Range<String> range = new Range<>((ObjectNode)rangeNode);
-            return new FeedRangeEpkImpl(range);
-        }
 
-        JsonNode pkNode = rootNode.get(Constants.Properties.FEED_RANGE_PARTITION_KEY);
-        if (pkNode != null && pkNode.isArray()) {
-            PartitionKeyInternal pk = mapper.convertValue(pkNode, PartitionKeyInternal.class);
-            return new FeedRangePartitionKeyImpl(pk);
-        }
-
-        JsonNode pkRangeIdNode =
-            rootNode.get(Constants.Properties.FEED_RANGE_PARTITION_KEY_RANGE_ID);
-        if (pkRangeIdNode != null && pkRangeIdNode.isTextual()) {
-            return new FeedRangePartitionKeyRangeImpl(pkRangeIdNode.asText());
+        try {
+            JsonNode rootNode = mapper.readTree(jsonString);
+
+            JsonNode rangeNode = rootNode.get(Constants.Properties.RANGE);
+            if (rangeNode != null && rangeNode.isObject()) {
+                Range<String> range = new Range<>((ObjectNode)rangeNode);
+                return new FeedRangeEpkImpl(range);
+            }
+
+            JsonNode pkNode = rootNode.get(Constants.Properties.FEED_RANGE_PARTITION_KEY);
+            if (pkNode != null && pkNode.isArray()) {
+                PartitionKeyInternal pk = mapper.convertValue(pkNode, PartitionKeyInternal.class);
+                return new FeedRangePartitionKeyImpl(pk);
+            }
+
+            JsonNode pkRangeIdNode =
+                rootNode.get(Constants.Properties.FEED_RANGE_PARTITION_KEY_RANGE_ID);
+            if (pkRangeIdNode != null && pkRangeIdNode.isTextual()) {
+                return new FeedRangePartitionKeyRangeImpl(pkRangeIdNode.asText());
+            }
+
+            return null;
+
+        } catch (final IOException ioError) {
+            LOGGER.debug("Failed to parse feed range JSON %s", jsonString, ioError);
+            return null;
         }
-
-        return null;
     }
 }