From 56be0d7f77b3d132846eb7c9d8a98de1ca3e52b5 Mon Sep 17 00:00:00 2001 From: David Noble Date: Tue, 17 Mar 2020 16:21:39 -0700 Subject: [PATCH] Completed port from v4. Unit tests pass locally. Next up: tests on cloud. --- .../data/cosmos/internal/RequestTimeline.java | 195 +++++ .../RntbdTransportClient.java | 88 ++- .../rntbd/RntbdClientChannelHandler.java | 6 +- .../RntbdClientChannelHealthChecker.java | 171 ++--- .../rntbd/RntbdConstants.java | 690 +++++++++++------- .../rntbd/RntbdContext.java | 14 +- .../rntbd/RntbdContextRequest.java | 4 +- .../rntbd/RntbdContextRequestEncoder.java | 36 +- .../rntbd/RntbdMetrics.java | 187 +++-- .../rntbd/RntbdRequestArgs.java | 44 +- .../rntbd/RntbdRequestDecoder.java | 16 +- .../rntbd/RntbdRequestEncoder.java | 23 +- .../rntbd/RntbdRequestFrame.java | 2 +- .../rntbd/RntbdRequestHeaders.java | 44 +- .../rntbd/RntbdRequestManager.java | 57 +- .../rntbd/RntbdRequestRecord.java | 187 ++++- .../rntbd/RntbdResponse.java | 245 ++++--- .../rntbd/RntbdResponseDecoder.java | 17 +- .../rntbd/RntbdServiceEndpoint.java | 16 +- .../directconnectivity/rntbd/RntbdToken.java | 38 +- .../rntbd/RntbdTokenStream.java | 58 +- .../rntbd/RntbdTokenType.java | 2 +- 22 files changed, 1413 insertions(+), 727 deletions(-) create mode 100644 sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/RequestTimeline.java diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/RequestTimeline.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/RequestTimeline.java new file mode 100644 index 0000000000000..acf582d960d30 --- /dev/null +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/RequestTimeline.java @@ -0,0 +1,195 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.data.cosmos.internal; + +import com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdObjectMapper; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonPropertyOrder; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.ser.std.ToStringSerializer; +import com.google.common.collect.ImmutableList; + +import java.time.Duration; +import java.time.OffsetDateTime; +import java.util.Iterator; + +import static com.google.common.base.Preconditions.checkNotNull; + +/** + * Represents the startTime and duration of important events in the lifetime of a request. + *

+ * A {@link RequestTimeline} represents a timeline as a sequence of {@link Event} instances with name, startTime, and + * duration properties. Hence, one might use this class to represent any timeline. Today we use it to represent + * request timelines for: + *

+ * A {@link RequestTimeline} serializes to JSON as an array of {@link Event} instances. This is the default + * serialization for any class that implements {@link Iterable}. + *

+ * Example: + * 

{@code OffsetDateTime startTime = OffsetDateTime.parse("2020-01-07T11:24:12.842749-08:00", DateTimeFormatter.ISO_OFFSET_DATE_TIME);
+ * sys.out.println(RequestTimeline.of(
+ *     new RequestTimeline.Event("foo", startTime, startTime.plusSeconds(1)),
+ *     new RequestTimeline.Event("bar", startTime.plusSeconds(1), startTime.plusSeconds(2))));}
+ * JSON serialization: + * 
{@code [{"name":"foo","startTime":"2020-01-07T11:24:12.842749-08:00","duration":"PT1S"},{"name":"bar","startTime":"2020-01-07T11:24:13.842749-08:00","duration":"PT1S"}])}
+ */ +public final class RequestTimeline implements Iterable { + + private static final RequestTimeline EMPTY = new RequestTimeline(); + private final ImmutableList events; + + private RequestTimeline() { + this.events = ImmutableList.of(); + } + + private RequestTimeline(final ImmutableList events) { + checkNotNull(events, "expected non-null events"); + this.events = events; + } + + /** + * Returns an empty {@link RequestTimeline}. + * + * The empty startTime line returned is static. + * + * @return an empty {@link RequestTimeline}. + */ + public static RequestTimeline empty() { + return EMPTY; + } + + /** + * Returns an iterator for enumerating the {@link Event} instances in this {@link RequestTimeline}. + * + * @return an iterator for enumerating the {@link Event} instances in this {@link RequestTimeline}. + */ + @Override + public Iterator iterator() { + return this.events.iterator(); + } + + /** + * Returns an empty {@link RequestTimeline}. + * + * The empty startTime line returned is static and equivalent to calling {@link RequestTimeline#empty}. + * + * @return an empty request timeline. + */ + public static RequestTimeline of() { + return EMPTY; + } + + /** + * Returns a new {@link RequestTimeline} with a single event. + * + * @return a new {@link RequestTimeline} with a single event. + */ + public static RequestTimeline of(final Event event) { + return new RequestTimeline(ImmutableList.of(event)); + } + + /** + * Returns a new {@link RequestTimeline} with a pair of events. + * + * @return a new {@link RequestTimeline} with a pair of events. + */ + public static RequestTimeline of(final Event e1, final Event e2) { + return new RequestTimeline(ImmutableList.of(e1, e2)); + } + + /** + * Returns a new {@link RequestTimeline} with three events. + * + * @return a new {@link RequestTimeline} with three events. + */ + public static RequestTimeline of(final Event e1, final Event e2, final Event e3) { + return new RequestTimeline(ImmutableList.of(e1, e2, e3)); + } + + /** + * Returns a new {@link RequestTimeline} with four events. + * + * @return a new {@link RequestTimeline} with four events. + */ + public static RequestTimeline of(final Event e1, final Event e2, final Event e3, final Event e4) { + return new RequestTimeline(ImmutableList.of(e1, e2, e3, e4)); + } + + /** + * Returns a new {@link RequestTimeline} with five events. + * + * @return a new {@link RequestTimeline} with five events. + */ + public static RequestTimeline of(final Event e1, final Event e2, final Event e3, final Event e4, final Event e5) { + return new RequestTimeline(ImmutableList.of(e1, e2, e3, e4, e5)); + } + + /** + * Returns a new {@link RequestTimeline} with an arbitrary number of events. + * + * @return a new {@link RequestTimeline} with an arbitrary number of events. + */ + public static RequestTimeline of(final Event... events) { + return new RequestTimeline(ImmutableList.copyOf(events)); + } + + /** + * Returns a textual representation of this {@link RequestTimeline}. + *

+ * The textual representation returned is a string of the form {@code RequestTimeline(} <event-array> + * {@code )}. + */ + @Override + public String toString() { + return RntbdObjectMapper.toString(this); + } + + @JsonPropertyOrder({ "name", "startTime", "durationInMicroSec" }) + public static final class Event { + + @JsonIgnore + private final Duration duration; + + @JsonSerialize(using = ToStringSerializer.class) + private final long durationInMicroSec; + + @JsonProperty("eventName") + private final String name; + + @JsonSerialize(using = ToStringSerializer.class) + private final OffsetDateTime startTime; + + public Event(final String name, final OffsetDateTime from, final OffsetDateTime to) { + + checkNotNull(name, "expected non-null name"); + + this.name = name; + this.startTime = from; + + this.duration = from == null ? null : to == null ? Duration.ZERO : Duration.between(from, to); + if(this.duration != null) { + this.durationInMicroSec = duration.toNanos()/1000L; + } else { + this.durationInMicroSec = 0; + } + } + + public Duration getDuration() { + return this.duration; + } + + public String getName() { + return name; + } + + public OffsetDateTime getStartTime() { + return startTime; + } + } +} diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/RntbdTransportClient.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/RntbdTransportClient.java index ea42e143ae13a..d2c89d73aaf77 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/RntbdTransportClient.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/RntbdTransportClient.java @@ -3,6 +3,7 @@ package com.azure.data.cosmos.internal.directconnectivity; +import com.azure.data.cosmos.BridgeInternal; import com.azure.data.cosmos.internal.Configs; import com.azure.data.cosmos.internal.RxDocumentServiceRequest; import com.azure.data.cosmos.internal.UserAgentContainer; @@ -23,7 +24,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.publisher.Mono; -import reactor.core.publisher.SignalType; import java.io.File; import java.io.IOException; @@ -122,14 +122,19 @@ public Mono invokeStoreAsync(final URI address, final RxDocumentS logger.debug("RntbdTransportClient.invokeStoreAsync({}, {}): {}", address, request, record); - return Mono.fromFuture(record).doFinally(signalType -> { - logger.debug("SignalType.{} received from reactor: {\n endpoint: {},\n record: {}\n}", - signalType.name(), - endpoint, - record); - if (signalType == SignalType.CANCEL) { - record.stage(RntbdRequestRecord.Stage.CANCELLED_BY_CLIENT); + return Mono.fromFuture(record.whenComplete((response, throwable) -> { + + record.stage(RntbdRequestRecord.Stage.COMPLETED); + + if (request.requestContext.cosmosResponseDiagnostics == null) { + request.requestContext.cosmosResponseDiagnostics = BridgeInternal.createCosmosResponseDiagnostics(); + } + + if(response != null) { + logger.debug("%s", record.takeTimelineSnapshot()); } + })).doOnCancel(() -> { + logger.debug("REQUEST CANCELLED: {}", record); }); } @@ -319,6 +324,48 @@ public String toString() { // region Types + /** + * A builder for constructing {@link Options} instances. + * + *

Using system properties to set the default {@link Options} used by an {@link Builder}

+ *

+ * A default options instance is created when the {@link Builder} class is initialized. This instance specifies + * the default options used by every {@link Builder} instance. In priority order the default options instance + * is created from: + *

    + *
  1. The JSON value of system property {@code azure.cosmos.directTcp.defaultOptions}. + *

    Example: + * 

    {@code -Dazure.cosmos.directTcp.defaultOptions={\"maxChannelsPerEndpoint\":5,\"maxRequestsPerChannel\":30}}
    + *
    2. + *
  2. The contents of the JSON file located by system property {@code azure.cosmos.directTcp + * .defaultOptionsFile}. + *

    Example: + * 

    {@code -Dazure.cosmos.directTcp.defaultOptionsFile=/path/to/default/options/file}
    + *
    3. + *
  3. The contents of JSON resource file {@code azure.cosmos.directTcp.defaultOptions.json}. + *

    Specifically, the resource file is read from this stream: + * 

    {@code RntbdTransportClient.class.getClassLoader().getResourceAsStream("azure.cosmos.directTcp.defaultOptions.json")}
    + *

    Example: 

    {@code {
+         *   "bufferPageSize": 8192,
+         *   "connectionTimeout": "PT1M",
+         *   "idleChannelTimeout": "PT0S",
+         *   "idleEndpointTimeout": "PT1M10S",
+         *   "maxBufferCapacity": 8388608,
+         *   "maxChannelsPerEndpoint": 10,
+         *   "maxRequestsPerChannel": 30,
+         *   "receiveHangDetectionTime": "PT1M5S",
+         *   "requestExpiryInterval": "PT5S",
+         *   "requestTimeout": "PT1M",
+         *   "requestTimerResolution": "PT0.5S",
+         *   "sendHangDetectionTime": "PT10S",
+         *   "shutdownTimeout": "PT15S"
+         * }}
    + *
    4. + *
+ *

JSON value errors are logged and then ignored. If none of the above values are available or all available + * values are in error, the default options instance is created from the private parameterless constructor for + * {@link Options}. + */ @SuppressWarnings("UnusedReturnValue") public static class Builder { @@ -329,16 +376,6 @@ public static class Builder { static { - // In priority order we take default Direct TCP options from: - // - // 1. the string value of system property "azure.cosmos.directTcp.options", or - // 2. the contents of the file located by the system property "azure.cosmos.directTcp.optionsFile", or - // 3. the contents of the resource file named "azure.cosmos.directTcp.options.json" - // - // Otherwise, if none of these values are set or an error occurs we create default options based on a - // set of hard-wired values defined in the default private parameterless constructor for - // RntbdTransportClient.Options. - Options options = null; try { @@ -373,7 +410,7 @@ public static class Builder { final ClassLoader loader = RntbdTransportClient.class.getClassLoader(); final String name = DEFAULT_OPTIONS_PROPERTY_NAME + ".json"; - try (final InputStream stream = loader.getResourceAsStream(name)) { + try (InputStream stream = loader.getResourceAsStream(name)) { if (stream != null) { // Attempt to load default options from the JSON resource file "{propertyName}.json" options = RntbdObjectMapper.readValue(stream, Options.class); @@ -383,7 +420,14 @@ public static class Builder { } } } finally { - DEFAULT_OPTIONS = options != null ? options : new Options(); + if (options == null) { + DEFAULT_OPTIONS = new Options(); + } else { + logger.info("Updated default Direct TCP options from system property {}: {}", + DEFAULT_OPTIONS_PROPERTY_NAME, + options); + DEFAULT_OPTIONS = options; + } } } @@ -553,7 +597,9 @@ public Builder userAgent(final UserAgentContainer value) { static final class JsonSerializer extends StdSerializer { - public JsonSerializer() { + private static final long serialVersionUID = 1007663695768825670L; + + JsonSerializer() { super(RntbdTransportClient.class); } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdClientChannelHandler.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdClientChannelHandler.java index bd5bcc38d3f7e..e56559ff3aed2 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdClientChannelHandler.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdClientChannelHandler.java @@ -44,7 +44,7 @@ public class RntbdClientChannelHandler extends ChannelInitializer imple */ @Override public void channelAcquired(final Channel channel) { - logger.trace("{} CHANNEL ACQUIRED", channel); + logger.debug("{} CHANNEL ACQUIRED", channel); } /** @@ -56,7 +56,7 @@ public void channelAcquired(final Channel channel) { */ @Override public void channelCreated(final Channel channel) { - logger.trace("{} CHANNEL CREATED", channel); + logger.debug("{} CHANNEL CREATED", channel); this.initChannel(channel); } @@ -69,7 +69,7 @@ public void channelCreated(final Channel channel) { */ @Override public void channelReleased(final Channel channel) { - logger.trace("{} CHANNEL RELEASED", channel); + logger.debug("{} CHANNEL RELEASED", channel); } /** diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdClientChannelHealthChecker.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdClientChannelHealthChecker.java index c1020f01988af..de354274a58a5 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdClientChannelHealthChecker.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdClientChannelHealthChecker.java @@ -4,10 +4,7 @@ package com.azure.data.cosmos.internal.directconnectivity.rntbd; import com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdEndpoint.Config; -import com.fasterxml.jackson.core.JsonGenerator; -import com.fasterxml.jackson.databind.SerializerProvider; -import com.fasterxml.jackson.databind.annotation.JsonSerialize; -import com.fasterxml.jackson.databind.ser.std.StdSerializer; +import com.fasterxml.jackson.annotation.JsonProperty; import io.netty.channel.Channel; import io.netty.channel.pool.ChannelHealthChecker; import io.netty.util.concurrent.Future; @@ -15,7 +12,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; import java.util.Optional; import java.util.concurrent.atomic.AtomicLongFieldUpdater; @@ -24,7 +20,6 @@ import static com.google.common.base.Preconditions.checkNotNull; import static java.util.concurrent.atomic.AtomicLongFieldUpdater.newUpdater; -@JsonSerialize(using = RntbdClientChannelHealthChecker.JsonSerializer.class) public final class RntbdClientChannelHealthChecker implements ChannelHealthChecker { // region Fields @@ -43,22 +38,17 @@ public final class RntbdClientChannelHealthChecker implements ChannelHealthCheck // A channel will not be declared unhealthy if a write was attempted recently. As such gaps between // Timestamps.lastChannelWriteAttempt and Timestamps.lastChannelWrite lower than this value are ignored. - // Guidance: The grace period should be large enough to accommodate slow writes. For example, a value of 2s requires - // that the client can sustain data rates of at least 1 MB/s when writing 2 MB documents. + // Guidance: The grace period should be large enough to accommodate slow writes. For example, a value of 2s + // requires that the client can sustain data rates of at least 1 MB/s when writing 2 MB documents. private static final long writeHangGracePeriodInNanos = 2L * 1_000_000_000L; - // A channel is considered idle if: - // idleConnectionTimeout > 0L && System.nanoTime() - Timestamps.lastChannelRead() >= idleConnectionTimeout + @JsonProperty private final long idleConnectionTimeoutInNanos; - // A channel will be declared unhealthy if the gap between Timestamps.lastChannelWrite and Timestamps.lastChannelRead - // grows beyond this value. - // Constraint: readDelayLimit > readHangGracePeriod + @JsonProperty private final long readDelayLimitInNanos; - // A channel will be declared unhealthy if the gap between Timestamps.lastChannelWriteAttempt and Timestamps.lastChannelWrite - // grows beyond this value. - // Constraint: writeDelayLimit > writeHangGracePeriod + @JsonProperty private final long writeDelayLimitInNanos; // endregion @@ -67,33 +57,72 @@ public final class RntbdClientChannelHealthChecker implements ChannelHealthCheck public RntbdClientChannelHealthChecker(final Config config) { - checkNotNull(config, "config: null"); + checkNotNull(config, "expected non-null config"); - this.idleConnectionTimeoutInNanos = config.idleConnectionTimeoutInNanos(); + checkArgument(config.receiveHangDetectionTimeInNanos() > readHangGracePeriodInNanos, + "config.receiveHangDetectionTimeInNanos: %s", + config.receiveHangDetectionTimeInNanos()); - this.readDelayLimitInNanos = config.receiveHangDetectionTimeInNanos(); - checkArgument(this.readDelayLimitInNanos > readHangGracePeriodInNanos, "config.receiveHangDetectionTimeInNanos: %s", this.readDelayLimitInNanos); + checkArgument(config.sendHangDetectionTimeInNanos() > writeHangGracePeriodInNanos, + "config.sendHangDetectionTimeInNanos: %s", + config.sendHangDetectionTimeInNanos()); + this.idleConnectionTimeoutInNanos = config.idleConnectionTimeoutInNanos(); + this.readDelayLimitInNanos = config.receiveHangDetectionTimeInNanos(); this.writeDelayLimitInNanos = config.sendHangDetectionTimeInNanos(); - checkArgument(this.writeDelayLimitInNanos > writeHangGracePeriodInNanos, "config.sendHangDetectionTimeInNanos: %s", this.writeDelayLimitInNanos); + } // endregion // region Methods + /** + * Returns the idle connection timeout interval in nanoseconds. + *

+ * A channel is considered idle if {@link #idleConnectionTimeoutInNanos} is greater than zero and the time since + * the last channel read is greater than {@link #idleConnectionTimeoutInNanos}. + * + * @return Idle connection timeout interval in nanoseconds. + */ public long idleConnectionTimeoutInNanos() { return this.idleConnectionTimeoutInNanos; } + /** + * Returns the read delay limit in nanoseconds. + *

+ * A channel will be declared unhealthy if the gap between the last channel write and the last channel read grows + * beyond this value. + *

+ * Constraint: {@link #readDelayLimitInNanos} > {@link #readHangGracePeriodInNanos} + * + * @return Read delay limit in nanoseconds. + */ public long readDelayLimitInNanos() { return this.readDelayLimitInNanos; } + /** + * Returns the write delay limit in nanoseconds. + *

+ * A channel will be declared unhealthy if the gap between the last channel write attempt and the last channel write + * grows beyond this value. + *

+ * Constraint: {@link #writeDelayLimitInNanos} > {@link #writeHangGracePeriodInNanos} + * + * @return Write delay limit in nanoseconds. + */ public long writeDelayLimitInNanos() { return this.writeDelayLimitInNanos; } + /** + * Determines whether a specified channel is healthy. + * + * @param channel A channel whose health is to be checked. + * @return A future with a result of {@code true} if the channel is healthy, or {@code false} otherwise. + */ public Future isHealthy(final Channel channel) { checkNotNull(channel, "expected non-null channel"); @@ -109,7 +138,7 @@ public Future isHealthy(final Channel channel) { final Timestamps timestamps = requestManager.snapshotTimestamps(); final long currentTime = System.nanoTime(); - if (currentTime - timestamps.lastChannelRead() < recentReadWindowInNanos) { + if (currentTime - timestamps.lastChannelReadNanoTime() < recentReadWindowInNanos) { return promise.setSuccess(Boolean.TRUE); // because we recently received data } @@ -117,17 +146,22 @@ public Future isHealthy(final Channel channel) { // Treat the channel as unhealthy if the gap between the last attempted write and the last successful write // grew beyond acceptable limits, unless a write was attempted recently. This is a sign of a hung write. - final long writeDelay = timestamps.lastChannelWriteAttempt() - timestamps.lastChannelWrite(); + final long writeDelayInNanos = + timestamps.lastChannelWriteAttemptNanoTime() - timestamps.lastChannelWriteNanoTime(); - if (writeDelay > this.writeDelayLimitInNanos && currentTime - timestamps.lastChannelWriteAttempt() > writeHangGracePeriodInNanos) { + final long writeHangDurationInNanos = + currentTime - timestamps.lastChannelWriteAttemptNanoTime(); + + if (writeDelayInNanos > this.writeDelayLimitInNanos && writeHangDurationInNanos > writeHangGracePeriodInNanos) { final Optional rntbdContext = requestManager.rntbdContext(); final int pendingRequestCount = requestManager.pendingRequestCount(); - logger.warn("{} health check failed due to hung write: {lastChannelWriteAttempt: {}, lastChannelWrite: {}, " - + "writeDelay: {}, writeDelayLimit: {}, rntbdContext: {}, pendingRequestCount: {}}", channel, - timestamps.lastChannelWriteAttempt(), timestamps.lastChannelWrite(), writeDelay, - this.writeDelayLimitInNanos, rntbdContext, pendingRequestCount); + logger.warn("{} health check failed due to hung write: {lastChannelWriteAttemptNanoTime: {}, " + + "lastChannelWriteNanoTime: {}, writeDelayInNanos: {}, writeDelayLimitInNanos: {}, " + + "rntbdContext: {}, pendingRequestCount: {}}", + channel, timestamps.lastChannelWriteAttemptNanoTime(), timestamps.lastChannelWriteNanoTime(), + writeDelayInNanos, this.writeDelayLimitInNanos, rntbdContext, pendingRequestCount); return promise.setSuccess(Boolean.FALSE); } @@ -136,23 +170,24 @@ public Future isHealthy(final Channel channel) { // Treat the connection as unhealthy if the gap between the last successful write and the last successful read // grew beyond acceptable limits, unless a write succeeded recently. This is a sign of a hung read. - final long readDelay = timestamps.lastChannelWrite() - timestamps.lastChannelRead(); + final long readDelay = timestamps.lastChannelWriteNanoTime() - timestamps.lastChannelReadNanoTime(); + final long readHangDuration = currentTime - timestamps.lastChannelWriteNanoTime(); - if (readDelay > this.readDelayLimitInNanos && currentTime - timestamps.lastChannelWrite() > readHangGracePeriodInNanos) { + if (readDelay > this.readDelayLimitInNanos && readHangDuration > readHangGracePeriodInNanos) { final Optional rntbdContext = requestManager.rntbdContext(); final int pendingRequestCount = requestManager.pendingRequestCount(); logger.warn("{} health check failed due to hung read: {lastChannelWrite: {}, lastChannelRead: {}, " + "readDelay: {}, readDelayLimit: {}, rntbdContext: {}, pendingRequestCount: {}}", channel, - timestamps.lastChannelWrite(), timestamps.lastChannelRead(), readDelay, + timestamps.lastChannelWriteNanoTime(), timestamps.lastChannelReadNanoTime(), readDelay, this.readDelayLimitInNanos, rntbdContext, pendingRequestCount); return promise.setSuccess(Boolean.FALSE); } if (this.idleConnectionTimeoutInNanos > 0L) { - if (currentTime - timestamps.lastChannelRead() > this.idleConnectionTimeoutInNanos) { + if (currentTime - timestamps.lastChannelReadNanoTime() > this.idleConnectionTimeoutInNanos) { return promise.setSuccess(Boolean.FALSE); } } @@ -178,41 +213,24 @@ public String toString() { // region Types - static final class JsonSerializer extends StdSerializer { - - JsonSerializer() { - super(RntbdClientChannelHealthChecker.class); - } - - @Override - public void serialize(RntbdClientChannelHealthChecker value, JsonGenerator generator, SerializerProvider provider) throws IOException { - generator.writeStartObject(); - generator.writeNumberField("idleConnectionTimeoutInNanos", value.idleConnectionTimeoutInNanos()); - generator.writeNumberField("readDelayLimitInNanos", value.readDelayLimitInNanos()); - generator.writeNumberField("writeDelayLimitInNanos", value.writeDelayLimitInNanos()); - generator.writeEndObject(); - } - } - - @JsonSerialize(using = Timestamps.JsonSerializer.class) static final class Timestamps { private static final AtomicLongFieldUpdater lastPingUpdater = - newUpdater(Timestamps.class, "lastPing"); + newUpdater(Timestamps.class, "lastPingNanoTime"); private static final AtomicLongFieldUpdater lastReadUpdater = - newUpdater(Timestamps.class, "lastRead"); + newUpdater(Timestamps.class, "lastReadNanoTime"); private static final AtomicLongFieldUpdater lastWriteUpdater = - newUpdater(Timestamps.class, "lastWrite"); + newUpdater(Timestamps.class, "lastWriteNanoTime"); private static final AtomicLongFieldUpdater lastWriteAttemptUpdater = - newUpdater(Timestamps.class, "lastWriteAttempt"); + newUpdater(Timestamps.class, "lastWriteAttemptNanoTime"); - private volatile long lastPing; - private volatile long lastRead; - private volatile long lastWrite; - private volatile long lastWriteAttempt; + private volatile long lastPingNanoTime; + private volatile long lastReadNanoTime; + private volatile long lastWriteNanoTime; + private volatile long lastWriteAttemptNanoTime; public Timestamps() { } @@ -220,10 +238,10 @@ public Timestamps() { @SuppressWarnings("CopyConstructorMissesField") public Timestamps(Timestamps other) { checkNotNull(other, "other: null"); - this.lastPing = lastPingUpdater.get(other); - this.lastRead = lastReadUpdater.get(other); - this.lastWrite = lastWriteUpdater.get(other); - this.lastWriteAttempt = lastWriteAttemptUpdater.get(other); + this.lastPingNanoTime = lastPingUpdater.get(other); + this.lastReadNanoTime = lastReadUpdater.get(other); + this.lastWriteNanoTime = lastWriteUpdater.get(other); + this.lastWriteAttemptNanoTime = lastWriteAttemptUpdater.get(other); } public void channelPingCompleted() { @@ -242,42 +260,29 @@ public void channelWriteCompleted() { lastWriteAttemptUpdater.set(this, System.nanoTime()); } - public long lastChannelPing() { + @JsonProperty + public long lastChannelPingNanoTime() { return lastPingUpdater.get(this); } - public long lastChannelRead() { + @JsonProperty + public long lastChannelReadNanoTime() { return lastReadUpdater.get(this); } - public long lastChannelWrite() { + @JsonProperty + public long lastChannelWriteNanoTime() { return lastWriteUpdater.get(this); } - public long lastChannelWriteAttempt() { + @JsonProperty + public long lastChannelWriteAttemptNanoTime() { return lastWriteAttemptUpdater.get(this); } @Override public String toString() { - return "RntbdClientChannelHealthChecker.Timestamps(" + RntbdObjectMapper.toJson(this) + ')'; - } - - static final class JsonSerializer extends StdSerializer { - - JsonSerializer() { - super(Timestamps.class); - } - - @Override - public void serialize(Timestamps value, JsonGenerator generator, SerializerProvider provider) throws IOException { - generator.writeStartObject(); - generator.writeNumberField("lastChannelPing", value.lastChannelPing()); - generator.writeNumberField("lastChannelRead", value.lastChannelRead()); - generator.writeNumberField("lastChannelWrite", value.lastChannelWrite()); - generator.writeNumberField("lastChannelWriteAttempt", value.lastChannelWriteAttempt()); - generator.writeEndObject(); - } + return RntbdObjectMapper.toString(this); } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdConstants.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdConstants.java index 956b0c3a184d1..839baec8e8534 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdConstants.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdConstants.java @@ -3,30 +3,34 @@ package com.azure.data.cosmos.internal.directconnectivity.rntbd; -import com.azure.data.cosmos.internal.guava27.Strings; +import com.azure.data.cosmos.internal.OperationType; +import com.azure.data.cosmos.internal.ResourceType; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Sets; +import io.netty.handler.codec.DecoderException; import java.util.EnumSet; import java.util.stream.Collector; -final class RntbdConstants { +import static com.azure.data.cosmos.internal.guava27.Strings.lenientFormat; - static final int CurrentProtocolVersion = 0x00000001; +public final class RntbdConstants { + + static final int CURRENT_PROTOCOL_VERSION = 0x00000001; private RntbdConstants() { } public enum RntbdConsistencyLevel { - Strong((byte)0x00), - BoundedStaleness((byte)0x01), - Session((byte)0x02), - Eventual((byte)0x03), - ConsistentPrefix((byte)0x04), + Strong((byte) 0x00), + BoundedStaleness((byte) 0x01), + Session((byte) 0x02), + Eventual((byte) 0x03), + ConsistentPrefix((byte) 0x04), - Invalid((byte)0xFF); + Invalid((byte) 0xFF); private final byte id; @@ -41,10 +45,10 @@ public byte id() { public enum RntbdContentSerializationFormat { - JsonText((byte)0x00), - CosmosBinary((byte)0x01), + JsonText((byte) 0x00), + CosmosBinary((byte) 0x01), - Invalid((byte)0xFF); + Invalid((byte) 0xFF); private final byte id; @@ -60,12 +64,12 @@ public byte id() { @SuppressWarnings("UnstableApiUsage") enum RntbdContextHeader implements RntbdHeader { - ProtocolVersion((short)0x0000, RntbdTokenType.ULong, false), - ClientVersion((short)0x0001, RntbdTokenType.SmallString, false), - ServerAgent((short)0x0002, RntbdTokenType.SmallString, true), - ServerVersion((short)0x0003, RntbdTokenType.SmallString, true), - IdleTimeoutInSeconds((short)0x0004, RntbdTokenType.ULong, false), - UnauthenticatedTimeoutInSeconds((short)0x0005, RntbdTokenType.ULong, false); + ProtocolVersion((short) 0x0000, RntbdTokenType.ULong, false), + ClientVersion((short) 0x0001, RntbdTokenType.SmallString, false), + ServerAgent((short) 0x0002, RntbdTokenType.SmallString, true), + ServerVersion((short) 0x0003, RntbdTokenType.SmallString, true), + IdleTimeoutInSeconds((short) 0x0004, RntbdTokenType.ULong, false), + UnauthenticatedTimeoutInSeconds((short) 0x0005, RntbdTokenType.ULong, false); public static final ImmutableMap map; public static final ImmutableSet set = Sets.immutableEnumSet(EnumSet.allOf(RntbdContextHeader.class)); @@ -100,9 +104,9 @@ public RntbdTokenType type() { enum RntbdContextRequestHeader implements RntbdHeader { - ProtocolVersion((short)0x0000, RntbdTokenType.ULong, true), - ClientVersion((short)0x0001, RntbdTokenType.SmallString, true), - UserAgent((short)0x0002, RntbdTokenType.SmallString, true); + ProtocolVersion((short) 0x0000, RntbdTokenType.ULong, true), + ClientVersion((short) 0x0001, RntbdTokenType.SmallString, true), + UserAgent((short) 0x0002, RntbdTokenType.SmallString, true); public static final ImmutableMap map; public static final ImmutableSet set = Sets.immutableEnumSet(EnumSet.allOf(RntbdContextRequestHeader.class)); @@ -137,10 +141,10 @@ public RntbdTokenType type() { public enum RntbdEnumerationDirection { - Invalid((byte)0x00), + Invalid((byte) 0x00), - Forward((byte)0x01), - Reverse((byte)0x02); + Forward((byte) 0x01), + Reverse((byte) 0x02); private final byte id; @@ -155,8 +159,8 @@ public byte id() { public enum RntbdFanoutOperationState { - Started((byte)0x01), - Completed((byte)0x02); + Started((byte) 0x01), + Completed((byte) 0x02); private final byte id; @@ -171,10 +175,10 @@ public byte id() { enum RntbdIndexingDirective { - Default((byte)0x00), - Include((byte)0x01), - Exclude((byte)0x02), - Invalid((byte)0xFF); + Default((byte) 0x00), + Include((byte) 0x01), + Exclude((byte) 0x02), + Invalid((byte) 0xFF); private final byte id; @@ -184,13 +188,13 @@ enum RntbdIndexingDirective { public static RntbdIndexingDirective fromId(final byte id) { switch (id) { - case (byte)0x00: + case (byte) 0x00: return Default; - case (byte)0x01: + case (byte) 0x01: return Include; - case (byte)0x02: + case (byte) 0x02: return Exclude; - case (byte)0xFF: + case (byte) 0xFF: return Invalid; } throw new IllegalArgumentException("id"); @@ -203,10 +207,10 @@ public byte id() { public enum RntbdMigrateCollectionDirective { - Thaw((byte)0x00), - Freeze((byte)0x01), + Thaw((byte) 0x00), + Freeze((byte) 0x01), - Invalid((byte)0xFF); + Invalid((byte) 0xFF); private final byte id; @@ -219,51 +223,53 @@ public byte id() { } } - enum RntbdOperationType { + public enum RntbdOperationType { - Connection((short)0x0000), - Create((short)0x0001), - Update((short)0x0002), - Read((short)0x0003), - ReadFeed((short)0x0004), - Delete((short)0x0005), - Replace((short)0x0006), + Connection((short) 0x0000, null), + Create((short) 0x0001, OperationType.Create), + Update((short) 0x0002, OperationType.Update), + Read((short) 0x0003, OperationType.Read), + ReadFeed((short) 0x0004, OperationType.ReadFeed), + Delete((short) 0x0005, OperationType.Delete), + Replace((short) 0x0006, OperationType.Replace), // Obsolete and now undefined: JPathQuery((short)0x0007), - ExecuteJavaScript((short)0x0008), - SQLQuery((short)0x0009), - Pause((short)0x000A), - Resume((short)0x000B), - Stop((short)0x000C), - Recycle((short)0x000D), - Crash((short)0x000E), - Query((short)0x000F), - ForceConfigRefresh((short)0x0010), - Head((short)0x0011), - HeadFeed((short)0x0012), - Upsert((short)0x0013), - Recreate((short)0x0014), - Throttle((short)0x0015), - GetSplitPoint((short)0x0016), - PreCreateValidation((short)0x0017), - BatchApply((short)0x0018), - AbortSplit((short)0x0019), - CompleteSplit((short)0x001A), - OfferUpdateOperation((short)0x001B), - OfferPreGrowValidation((short)0x001C), - BatchReportThroughputUtilization((short)0x001D), - CompletePartitionMigration((short)0x001E), - AbortPartitionMigration((short)0x001F), - PreReplaceValidation((short)0x0020), - AddComputeGatewayRequestCharges((short)0x0021), - MigratePartition((short)0x0022); + ExecuteJavaScript((short) 0x0008, OperationType.ExecuteJavaScript), + SqlQuery((short) 0x0009, OperationType.SqlQuery), + Pause((short) 0x000A, OperationType.Pause), + Resume((short) 0x000B, OperationType.Resume), + Stop((short) 0x000C, OperationType.Stop), + Recycle((short) 0x000D, OperationType.Recycle), + Crash((short) 0x000E, OperationType.Crash), + Query((short) 0x000F, OperationType.Query), + ForceConfigRefresh((short) 0x0010, OperationType.ForceConfigRefresh), + Head((short) 0x0011, OperationType.Head), + HeadFeed((short) 0x0012, OperationType.HeadFeed), + Upsert((short) 0x0013, OperationType.Upsert), + Recreate((short) 0x0014, OperationType.Recreate), + Throttle((short) 0x0015, OperationType.Throttle), + GetSplitPoint((short) 0x0016, OperationType.GetSplitPoint), + PreCreateValidation((short) 0x0017, OperationType.PreCreateValidation), + BatchApply((short) 0x0018, OperationType.BatchApply), + AbortSplit((short) 0x0019, OperationType.AbortSplit), + CompleteSplit((short) 0x001A, OperationType.CompleteSplit), + OfferUpdateOperation((short) 0x001B, OperationType.OfferUpdateOperation), + OfferPreGrowValidation((short) 0x001C, OperationType.OfferPreGrowValidation), + BatchReportThroughputUtilization((short) 0x001D, OperationType.BatchReportThroughputUtilization), + CompletePartitionMigration((short) 0x001E, OperationType.CompletePartitionMigration), + AbortPartitionMigration((short) 0x001F, OperationType.AbortPartitionMigration), + PreReplaceValidation((short) 0x0020, OperationType.PreReplaceValidation), + AddComputeGatewayRequestCharges((short) 0x0021, OperationType.AddComputeGatewayRequestCharges), + MigratePartition((short) 0x0022, OperationType.MigratePartition); private final short id; + private final OperationType type; - RntbdOperationType(final short id) { + RntbdOperationType(final short id, final OperationType type) { this.id = id; + this.type = type; } - public static RntbdOperationType fromId(final short id) throws IllegalArgumentException { + public static RntbdOperationType fromId(final short id) { switch (id) { case 0x0000: @@ -284,7 +290,7 @@ public static RntbdOperationType fromId(final short id) throws IllegalArgumentEx case 0x0008: return RntbdOperationType.ExecuteJavaScript; case 0x0009: - return RntbdOperationType.SQLQuery; + return RntbdOperationType.SqlQuery; case 0x000A: return RntbdOperationType.Pause; case 0x000B: @@ -335,20 +341,100 @@ public static RntbdOperationType fromId(final short id) throws IllegalArgumentEx return RntbdOperationType.AddComputeGatewayRequestCharges; case 0x0022: return RntbdOperationType.MigratePartition; + default: + throw new DecoderException(lenientFormat("expected byte value matching %s value, not %s", + RntbdOperationType.class.getSimpleName(), + id)); + } + } + + public static RntbdOperationType fromType(OperationType type) { + switch (type) { + case Crash: + return RntbdOperationType.Crash; + case Create: + return RntbdOperationType.Create; + case Delete: + return RntbdOperationType.Delete; + case ExecuteJavaScript: + return RntbdOperationType.ExecuteJavaScript; + case Query: + return RntbdOperationType.Query; + case Pause: + return RntbdOperationType.Pause; + case Read: + return RntbdOperationType.Read; + case ReadFeed: + return RntbdOperationType.ReadFeed; + case Recreate: + return RntbdOperationType.Recreate; + case Recycle: + return RntbdOperationType.Recycle; + case Replace: + return RntbdOperationType.Replace; + case Resume: + return RntbdOperationType.Resume; + case Stop: + return RntbdOperationType.Stop; + case SqlQuery: + return RntbdOperationType.SqlQuery; + case Update: + return RntbdOperationType.Update; + case ForceConfigRefresh: + return RntbdOperationType.ForceConfigRefresh; + case Head: + return RntbdOperationType.Head; + case HeadFeed: + return RntbdOperationType.HeadFeed; + case Upsert: + return RntbdOperationType.Upsert; + case Throttle: + return RntbdOperationType.Throttle; + case PreCreateValidation: + return RntbdOperationType.PreCreateValidation; + case GetSplitPoint: + return RntbdOperationType.GetSplitPoint; + case AbortSplit: + return RntbdOperationType.AbortSplit; + case CompleteSplit: + return RntbdOperationType.CompleteSplit; + case BatchApply: + return RntbdOperationType.BatchApply; + case OfferUpdateOperation: + return RntbdOperationType.OfferUpdateOperation; + case OfferPreGrowValidation: + return RntbdOperationType.OfferPreGrowValidation; + case BatchReportThroughputUtilization: + return RntbdOperationType.BatchReportThroughputUtilization; + case AbortPartitionMigration: + return RntbdOperationType.AbortPartitionMigration; + case CompletePartitionMigration: + return RntbdOperationType.CompletePartitionMigration; + case PreReplaceValidation: + return RntbdOperationType.PreReplaceValidation; + case MigratePartition: + return RntbdOperationType.MigratePartition; + case AddComputeGatewayRequestCharges: + return RntbdOperationType.AddComputeGatewayRequestCharges; + default: + throw new IllegalArgumentException(lenientFormat("unrecognized operation type: %s", type)); } - throw new IllegalArgumentException("id"); } public short id() { return this.id; } + + public OperationType type() { + return this.type; + } } public enum RntbdReadFeedKeyType { - Invalid((byte)0x00), - ResourceId((byte)0x01), - EffectivePartitionKey((byte)0x02); + Invalid((byte) 0x00), + ResourceId((byte) 0x01), + EffectivePartitionKey((byte) 0x02); private final byte id; @@ -363,10 +449,10 @@ public byte id() { public enum RntbdRemoteStorageType { - Invalid((byte)0x00), - NotSpecified((byte)0x01), - Standard((byte)0x02), - Premium((byte)0x03); + Invalid((byte) 0x00), + NotSpecified((byte) 0x01), + Standard((byte) 0x02), + Premium((byte) 0x03); private final byte id; @@ -381,111 +467,111 @@ public byte id() { public enum RntbdRequestHeader implements RntbdHeader { - ResourceId((short)0x0000, RntbdTokenType.Bytes, false), - AuthorizationToken((short)0x0001, RntbdTokenType.String, false), - PayloadPresent((short)0x0002, RntbdTokenType.Byte, true), - Date((short)0x0003, RntbdTokenType.SmallString, false), - PageSize((short)0x0004, RntbdTokenType.ULong, false), - SessionToken((short)0x0005, RntbdTokenType.String, false), - ContinuationToken((short)0x0006, RntbdTokenType.String, false), - IndexingDirective((short)0x0007, RntbdTokenType.Byte, false), - Match((short)0x0008, RntbdTokenType.String, false), - PreTriggerInclude((short)0x0009, RntbdTokenType.String, false), - PostTriggerInclude((short)0x000A, RntbdTokenType.String, false), - IsFanout((short)0x000B, RntbdTokenType.Byte, false), - CollectionPartitionIndex((short)0x000C, RntbdTokenType.ULong, false), - CollectionServiceIndex((short)0x000D, RntbdTokenType.ULong, false), - PreTriggerExclude((short)0x000E, RntbdTokenType.String, false), - PostTriggerExclude((short)0x000F, RntbdTokenType.String, false), - ConsistencyLevel((short)0x0010, RntbdTokenType.Byte, false), - EntityId((short)0x0011, RntbdTokenType.String, false), - ResourceSchemaName((short)0x0012, RntbdTokenType.SmallString, false), - ReplicaPath((short)0x0013, RntbdTokenType.String, true), - ResourceTokenExpiry((short)0x0014, RntbdTokenType.ULong, false), - DatabaseName((short)0x0015, RntbdTokenType.String, false), - CollectionName((short)0x0016, RntbdTokenType.String, false), - DocumentName((short)0x0017, RntbdTokenType.String, false), - AttachmentName((short)0x0018, RntbdTokenType.String, false), - UserName((short)0x0019, RntbdTokenType.String, false), - PermissionName((short)0x001A, RntbdTokenType.String, false), - StoredProcedureName((short)0x001B, RntbdTokenType.String, false), - UserDefinedFunctionName((short)0x001C, RntbdTokenType.String, false), - TriggerName((short)0x001D, RntbdTokenType.String, false), - EnableScanInQuery((short)0x001E, RntbdTokenType.Byte, false), - EmitVerboseTracesInQuery((short)0x001F, RntbdTokenType.Byte, false), - ConflictName((short)0x0020, RntbdTokenType.String, false), - BindReplicaDirective((short)0x0021, RntbdTokenType.String, false), - PrimaryMasterKey((short)0x0022, RntbdTokenType.String, false), - SecondaryMasterKey((short)0x0023, RntbdTokenType.String, false), - PrimaryReadonlyKey((short)0x0024, RntbdTokenType.String, false), - SecondaryReadonlyKey((short)0x0025, RntbdTokenType.String, false), - ProfileRequest((short)0x0026, RntbdTokenType.Byte, false), - EnableLowPrecisionOrderBy((short)0x0027, RntbdTokenType.Byte, false), - ClientVersion((short)0x0028, RntbdTokenType.SmallString, false), - CanCharge((short)0x0029, RntbdTokenType.Byte, false), - CanThrottle((short)0x002A, RntbdTokenType.Byte, false), - PartitionKey((short)0x002B, RntbdTokenType.String, false), - PartitionKeyRangeId((short)0x002C, RntbdTokenType.String, false), - NotUsed2D((short)0x002D, RntbdTokenType.Invalid, false), - NotUsed2E((short)0x002E, RntbdTokenType.Invalid, false), - NotUsed2F((short)0x002F, RntbdTokenType.Invalid, false), + ResourceId((short) 0x0000, RntbdTokenType.Bytes, false), + AuthorizationToken((short) 0x0001, RntbdTokenType.String, false), + PayloadPresent((short) 0x0002, RntbdTokenType.Byte, true), + Date((short) 0x0003, RntbdTokenType.SmallString, false), + PageSize((short) 0x0004, RntbdTokenType.ULong, false), + SessionToken((short) 0x0005, RntbdTokenType.String, false), + ContinuationToken((short) 0x0006, RntbdTokenType.String, false), + IndexingDirective((short) 0x0007, RntbdTokenType.Byte, false), + Match((short) 0x0008, RntbdTokenType.String, false), + PreTriggerInclude((short) 0x0009, RntbdTokenType.String, false), + PostTriggerInclude((short) 0x000A, RntbdTokenType.String, false), + IsFanout((short) 0x000B, RntbdTokenType.Byte, false), + CollectionPartitionIndex((short) 0x000C, RntbdTokenType.ULong, false), + CollectionServiceIndex((short) 0x000D, RntbdTokenType.ULong, false), + PreTriggerExclude((short) 0x000E, RntbdTokenType.String, false), + PostTriggerExclude((short) 0x000F, RntbdTokenType.String, false), + ConsistencyLevel((short) 0x0010, RntbdTokenType.Byte, false), + EntityId((short) 0x0011, RntbdTokenType.String, false), + ResourceSchemaName((short) 0x0012, RntbdTokenType.SmallString, false), + ReplicaPath((short) 0x0013, RntbdTokenType.String, true), + ResourceTokenExpiry((short) 0x0014, RntbdTokenType.ULong, false), + DatabaseName((short) 0x0015, RntbdTokenType.String, false), + CollectionName((short) 0x0016, RntbdTokenType.String, false), + DocumentName((short) 0x0017, RntbdTokenType.String, false), + AttachmentName((short) 0x0018, RntbdTokenType.String, false), + UserName((short) 0x0019, RntbdTokenType.String, false), + PermissionName((short) 0x001A, RntbdTokenType.String, false), + StoredProcedureName((short) 0x001B, RntbdTokenType.String, false), + UserDefinedFunctionName((short) 0x001C, RntbdTokenType.String, false), + TriggerName((short) 0x001D, RntbdTokenType.String, false), + EnableScanInQuery((short) 0x001E, RntbdTokenType.Byte, false), + EmitVerboseTracesInQuery((short) 0x001F, RntbdTokenType.Byte, false), + ConflictName((short) 0x0020, RntbdTokenType.String, false), + BindReplicaDirective((short) 0x0021, RntbdTokenType.String, false), + PrimaryMasterKey((short) 0x0022, RntbdTokenType.String, false), + SecondaryMasterKey((short) 0x0023, RntbdTokenType.String, false), + PrimaryReadonlyKey((short) 0x0024, RntbdTokenType.String, false), + SecondaryReadonlyKey((short) 0x0025, RntbdTokenType.String, false), + ProfileRequest((short) 0x0026, RntbdTokenType.Byte, false), + EnableLowPrecisionOrderBy((short) 0x0027, RntbdTokenType.Byte, false), + ClientVersion((short) 0x0028, RntbdTokenType.SmallString, false), + CanCharge((short) 0x0029, RntbdTokenType.Byte, false), + CanThrottle((short) 0x002A, RntbdTokenType.Byte, false), + PartitionKey((short) 0x002B, RntbdTokenType.String, false), + PartitionKeyRangeId((short) 0x002C, RntbdTokenType.String, false), + NotUsed2D((short) 0x002D, RntbdTokenType.Invalid, false), + NotUsed2E((short) 0x002E, RntbdTokenType.Invalid, false), + NotUsed2F((short) 0x002F, RntbdTokenType.Invalid, false), // not used 0x0030, - MigrateCollectionDirective((short)0x0031, RntbdTokenType.Byte, false), - NotUsed32((short)0x0032, RntbdTokenType.Invalid, false), - SupportSpatialLegacyCoordinates((short)0x0033, RntbdTokenType.Byte, false), - PartitionCount((short)0x0034, RntbdTokenType.ULong, false), - CollectionRid((short)0x0035, RntbdTokenType.String, false), - PartitionKeyRangeName((short)0x0036, RntbdTokenType.String, false), + MigrateCollectionDirective((short) 0x0031, RntbdTokenType.Byte, false), + NotUsed32((short) 0x0032, RntbdTokenType.Invalid, false), + SupportSpatialLegacyCoordinates((short) 0x0033, RntbdTokenType.Byte, false), + PartitionCount((short) 0x0034, RntbdTokenType.ULong, false), + CollectionRid((short) 0x0035, RntbdTokenType.String, false), + PartitionKeyRangeName((short) 0x0036, RntbdTokenType.String, false), // not used((short)0x0037), RoundTripTimeInMsec // not used((short)0x0038), RequestMessageSentTime // not used((short)0x0039), RequestMessageTimeOffset - SchemaName((short)0x003A, RntbdTokenType.String, false), - FilterBySchemaRid((short)0x003B, RntbdTokenType.String, false), - UsePolygonsSmallerThanAHemisphere((short)0x003C, RntbdTokenType.Byte, false), - GatewaySignature((short)0x003D, RntbdTokenType.String, false), - EnableLogging((short)0x003E, RntbdTokenType.Byte, false), - A_IM((short)0x003F, RntbdTokenType.String, false), - PopulateQuotaInfo((short)0x0040, RntbdTokenType.Byte, false), - DisableRUPerMinuteUsage((short)0x0041, RntbdTokenType.Byte, false), - PopulateQueryMetrics((short)0x0042, RntbdTokenType.Byte, false), - ResponseContinuationTokenLimitInKb((short)0x0043, RntbdTokenType.ULong, false), - PopulatePartitionStatistics((short)0x0044, RntbdTokenType.Byte, false), - RemoteStorageType((short)0x0045, RntbdTokenType.Byte, false), - CollectionRemoteStorageSecurityIdentifier((short)0x0046, RntbdTokenType.String, false), - IfModifiedSince((short)0x0047, RntbdTokenType.String, false), - PopulateCollectionThroughputInfo((short)0x0048, RntbdTokenType.Byte, false), - RemainingTimeInMsOnClientRequest((short)0x0049, RntbdTokenType.ULong, false), - ClientRetryAttemptCount((short)0x004A, RntbdTokenType.ULong, false), - TargetLsn((short)0x004B, RntbdTokenType.LongLong, false), - TargetGlobalCommittedLsn((short)0x004C, RntbdTokenType.LongLong, false), - TransportRequestID((short)0x004D, RntbdTokenType.ULong, false), - RestoreMetadaFilter((short)0x004E, RntbdTokenType.String, false), - RestoreParams((short)0x004F, RntbdTokenType.String, false), - ShareThroughput((short)0x0050, RntbdTokenType.Byte, false), - PartitionResourceFilter((short)0x0051, RntbdTokenType.String, false), - IsReadOnlyScript((short)0x0052, RntbdTokenType.Byte, false), - IsAutoScaleRequest((short)0x0053, RntbdTokenType.Byte, false), - ForceQueryScan((short)0x0054, RntbdTokenType.Byte, false), + SchemaName((short) 0x003A, RntbdTokenType.String, false), + FilterBySchemaRid((short) 0x003B, RntbdTokenType.String, false), + UsePolygonsSmallerThanAHemisphere((short) 0x003C, RntbdTokenType.Byte, false), + GatewaySignature((short) 0x003D, RntbdTokenType.String, false), + EnableLogging((short) 0x003E, RntbdTokenType.Byte, false), + A_IM((short) 0x003F, RntbdTokenType.String, false), + PopulateQuotaInfo((short) 0x0040, RntbdTokenType.Byte, false), + DisableRUPerMinuteUsage((short) 0x0041, RntbdTokenType.Byte, false), + PopulateQueryMetrics((short) 0x0042, RntbdTokenType.Byte, false), + ResponseContinuationTokenLimitInKb((short) 0x0043, RntbdTokenType.ULong, false), + PopulatePartitionStatistics((short) 0x0044, RntbdTokenType.Byte, false), + RemoteStorageType((short) 0x0045, RntbdTokenType.Byte, false), + CollectionRemoteStorageSecurityIdentifier((short) 0x0046, RntbdTokenType.String, false), + IfModifiedSince((short) 0x0047, RntbdTokenType.String, false), + PopulateCollectionThroughputInfo((short) 0x0048, RntbdTokenType.Byte, false), + RemainingTimeInMsOnClientRequest((short) 0x0049, RntbdTokenType.ULong, false), + ClientRetryAttemptCount((short) 0x004A, RntbdTokenType.ULong, false), + TargetLsn((short) 0x004B, RntbdTokenType.LongLong, false), + TargetGlobalCommittedLsn((short) 0x004C, RntbdTokenType.LongLong, false), + TransportRequestID((short) 0x004D, RntbdTokenType.ULong, false), + RestoreMetadaFilter((short) 0x004E, RntbdTokenType.String, false), + RestoreParams((short) 0x004F, RntbdTokenType.String, false), + ShareThroughput((short) 0x0050, RntbdTokenType.Byte, false), + PartitionResourceFilter((short) 0x0051, RntbdTokenType.String, false), + IsReadOnlyScript((short) 0x0052, RntbdTokenType.Byte, false), + IsAutoScaleRequest((short) 0x0053, RntbdTokenType.Byte, false), + ForceQueryScan((short) 0x0054, RntbdTokenType.Byte, false), // not used((short)0x0055), LeaseSeqNumber - CanOfferReplaceComplete((short)0x0056, RntbdTokenType.Byte, false), - ExcludeSystemProperties((short)0x0057, RntbdTokenType.Byte, false), - BinaryId((short)0x0058, RntbdTokenType.Bytes, false), - TimeToLiveInSeconds((short)0x0059, RntbdTokenType.Long, false), - EffectivePartitionKey((short)0x005A, RntbdTokenType.Bytes, false), - BinaryPassthroughRequest((short)0x005B, RntbdTokenType.Byte, false), - UserDefinedTypeName((short)0x005C, RntbdTokenType.String, false), - EnableDynamicRidRangeAllocation((short)0x005D, RntbdTokenType.Byte, false), - EnumerationDirection((short)0x005E, RntbdTokenType.Byte, false), - StartId((short)0x005F, RntbdTokenType.Bytes, false), - EndId((short)0x0060, RntbdTokenType.Bytes, false), - FanoutOperationState((short)0x0061, RntbdTokenType.Byte, false), - StartEpk((short)0x0062, RntbdTokenType.Bytes, false), - EndEpk((short)0x0063, RntbdTokenType.Bytes, false), - ReadFeedKeyType((short)0x0064, RntbdTokenType.Byte, false), - ContentSerializationFormat((short)0x0065, RntbdTokenType.Byte, false), - AllowTentativeWrites((short)0x0066, RntbdTokenType.Byte, false), - IsUserRequest((short)0x0067, RntbdTokenType.Byte, false), - SharedOfferThroughput((short)0x0068, RntbdTokenType.ULong, false); + CanOfferReplaceComplete((short) 0x0056, RntbdTokenType.Byte, false), + ExcludeSystemProperties((short) 0x0057, RntbdTokenType.Byte, false), + BinaryId((short) 0x0058, RntbdTokenType.Bytes, false), + TimeToLiveInSeconds((short) 0x0059, RntbdTokenType.Long, false), + EffectivePartitionKey((short) 0x005A, RntbdTokenType.Bytes, false), + BinaryPassthroughRequest((short) 0x005B, RntbdTokenType.Byte, false), + UserDefinedTypeName((short) 0x005C, RntbdTokenType.String, false), + EnableDynamicRidRangeAllocation((short) 0x005D, RntbdTokenType.Byte, false), + EnumerationDirection((short) 0x005E, RntbdTokenType.Byte, false), + StartId((short) 0x005F, RntbdTokenType.Bytes, false), + EndId((short) 0x0060, RntbdTokenType.Bytes, false), + FanoutOperationState((short) 0x0061, RntbdTokenType.Byte, false), + StartEpk((short) 0x0062, RntbdTokenType.Bytes, false), + EndEpk((short) 0x0063, RntbdTokenType.Bytes, false), + ReadFeedKeyType((short) 0x0064, RntbdTokenType.Byte, false), + ContentSerializationFormat((short) 0x0065, RntbdTokenType.Byte, false), + AllowTentativeWrites((short) 0x0066, RntbdTokenType.Byte, false), + IsUserRequest((short) 0x0067, RntbdTokenType.Byte, false), + SharedOfferThroughput((short) 0x0068, RntbdTokenType.ULong, false); public static final ImmutableMap map; public static final ImmutableSet set = Sets.immutableEnumSet(EnumSet.allOf(RntbdRequestHeader.class)); @@ -518,43 +604,45 @@ public RntbdTokenType type() { } } - enum RntbdResourceType { - - Connection((short)0x0000), - Database((short)0x0001), - Collection((short)0x0002), - Document((short)0x0003), - Attachment((short)0x0004), - User((short)0x0005), - Permission((short)0x0006), - StoredProcedure((short)0x0007), - Conflict((short)0x0008), - Trigger((short)0x0009), - UserDefinedFunction((short)0x000A), - Module((short)0x000B), - Replica((short)0x000C), - ModuleCommand((short)0x000D), - Record((short)0x000E), - Offer((short)0x000F), - PartitionSetInformation((short)0x0010), - XPReplicatorAddress((short)0x0011), - MasterPartition((short)0x0012), - ServerPartition((short)0x0013), - DatabaseAccount((short)0x0014), - Topology((short)0x0015), - PartitionKeyRange((short)0x0016), + public enum RntbdResourceType { + + Connection((short) 0x0000, null), + Database((short) 0x0001, ResourceType.Database), + Collection((short) 0x0002, ResourceType.DocumentCollection), + Document((short) 0x0003, ResourceType.Document), + Attachment((short) 0x0004, ResourceType.Attachment), + User((short) 0x0005, ResourceType.User), + Permission((short) 0x0006, ResourceType.Permission), + StoredProcedure((short) 0x0007, ResourceType.StoredProcedure), + Conflict((short) 0x0008, ResourceType.Conflict), + Trigger((short) 0x0009, ResourceType.Trigger), + UserDefinedFunction((short) 0x000A, ResourceType.UserDefinedFunction), + Module((short) 0x000B, ResourceType.Module), + Replica((short) 0x000C, ResourceType.Replica), + ModuleCommand((short) 0x000D, ResourceType.ModuleCommand), + Record((short) 0x000E, ResourceType.Record), + Offer((short) 0x000F, ResourceType.Offer), + PartitionSetInformation((short) 0x0010, ResourceType.PartitionSetInformation), + XPReplicatorAddress((short) 0x0011, ResourceType.XPReplicatorAddress), + MasterPartition((short) 0x0012, ResourceType.MasterPartition), + ServerPartition((short) 0x0013, ResourceType.ServerPartition), + DatabaseAccount((short) 0x0014, ResourceType.DatabaseAccount), + Topology((short) 0x0015, ResourceType.Topology), + PartitionKeyRange((short) 0x0016, ResourceType.PartitionKeyRange), // Obsolete and now undefined: Timestamp((short)0x0017), - Schema((short)0x0018), - BatchApply((short)0x0019), - RestoreMetadata((short)0x001A), - ComputeGatewayCharges((short)0x001B), - RidRange((short)0x001C), - UserDefinedType((short)0x001D); + Schema((short) 0x0018, ResourceType.Schema), + BatchApply((short) 0x0019, ResourceType.BatchApply), + RestoreMetadata((short) 0x001A, ResourceType.RestoreMetadata), + ComputeGatewayCharges((short) 0x001B, ResourceType.ComputeGatewayCharges), + RidRange((short) 0x001C, ResourceType.RidRange), + UserDefinedType((short) 0x001D, ResourceType.UserDefinedType); private final short id; + private final ResourceType type; - RntbdResourceType(final short id) { + RntbdResourceType(final short id, final ResourceType type) { this.id = id; + this.type = type; } public static RntbdResourceType fromId(final short id) throws IllegalArgumentException { @@ -605,7 +693,6 @@ public static RntbdResourceType fromId(final short id) throws IllegalArgumentExc return RntbdResourceType.Topology; case 0x0016: return RntbdResourceType.PartitionKeyRange; - // Obsolete and now undefined: case 0x0017: return RntbdResourceType.Timestamp; case 0x0018: return RntbdResourceType.Schema; case 0x0019: @@ -618,74 +705,145 @@ public static RntbdResourceType fromId(final short id) throws IllegalArgumentExc return RntbdResourceType.RidRange; case 0x001D: return RntbdResourceType.UserDefinedType; + default: + throw new DecoderException(lenientFormat("expected byte value matching %s value, not %s", + RntbdResourceType.class.getSimpleName(), + id)); + } + } + + public static RntbdResourceType fromType(ResourceType type) { + + switch (type) { + case Database: + return RntbdResourceType.Database; + case DocumentCollection: + return RntbdResourceType.Collection; + case Document: + return RntbdResourceType.Document; + case Attachment: + return RntbdResourceType.Attachment; + case User: + return RntbdResourceType.User; + case Permission: + return RntbdResourceType.Permission; + case StoredProcedure: + return RntbdResourceType.StoredProcedure; + case Conflict: + return RntbdResourceType.Conflict; + case Trigger: + return RntbdResourceType.Trigger; + case UserDefinedFunction: + return RntbdResourceType.UserDefinedFunction; + case Module: + return RntbdResourceType.Module; + case Replica: + return RntbdResourceType.Replica; + case ModuleCommand: + return RntbdResourceType.ModuleCommand; + case Record: + return RntbdResourceType.Record; + case Offer: + return RntbdResourceType.Offer; + case PartitionSetInformation: + return RntbdResourceType.PartitionSetInformation; + case XPReplicatorAddress: + return RntbdResourceType.XPReplicatorAddress; + case MasterPartition: + return RntbdResourceType.MasterPartition; + case ServerPartition: + return RntbdResourceType.ServerPartition; + case DatabaseAccount: + return RntbdResourceType.DatabaseAccount; + case Topology: + return RntbdResourceType.Topology; + case PartitionKeyRange: + return RntbdResourceType.PartitionKeyRange; + case Schema: + return RntbdResourceType.Schema; + case BatchApply: + return RntbdResourceType.BatchApply; + case RestoreMetadata: + return RntbdResourceType.RestoreMetadata; + case ComputeGatewayCharges: + return RntbdResourceType.ComputeGatewayCharges; + case RidRange: + return RntbdResourceType.RidRange; + case UserDefinedType: + return RntbdResourceType.UserDefinedType; + default: + throw new IllegalArgumentException(lenientFormat("unrecognized resource type: %s", type)); } - throw new IllegalArgumentException(Strings.lenientFormat("id: %s", id)); } public short id() { return this.id; } + + public ResourceType type() { + return this.type; + } } public enum RntbdResponseHeader implements RntbdHeader { - PayloadPresent((short)0x0000, RntbdTokenType.Byte, true), + PayloadPresent((short) 0x0000, RntbdTokenType.Byte, true), // not used((short)0x0001), - LastStateChangeDateTime((short)0x0002, RntbdTokenType.SmallString, false), - ContinuationToken((short)0x0003, RntbdTokenType.String, false), - ETag((short)0x0004, RntbdTokenType.String, false), + LastStateChangeDateTime((short) 0x0002, RntbdTokenType.SmallString, false), + ContinuationToken((short) 0x0003, RntbdTokenType.String, false), + ETag((short) 0x0004, RntbdTokenType.String, false), // not used((short)0x005,) // not used((short)0x006,) - ReadsPerformed((short)0x0007, RntbdTokenType.ULong, false), - WritesPerformed((short)0x0008, RntbdTokenType.ULong, false), - QueriesPerformed((short)0x0009, RntbdTokenType.ULong, false), - IndexTermsGenerated((short)0x000A, RntbdTokenType.ULong, false), - ScriptsExecuted((short)0x000B, RntbdTokenType.ULong, false), - RetryAfterMilliseconds((short)0x000C, RntbdTokenType.ULong, false), - IndexingDirective((short)0x000D, RntbdTokenType.Byte, false), - StorageMaxResoureQuota((short)0x000E, RntbdTokenType.String, false), - StorageResourceQuotaUsage((short)0x000F, RntbdTokenType.String, false), - SchemaVersion((short)0x0010, RntbdTokenType.SmallString, false), - CollectionPartitionIndex((short)0x0011, RntbdTokenType.ULong, false), - CollectionServiceIndex((short)0x0012, RntbdTokenType.ULong, false), - LSN((short)0x0013, RntbdTokenType.LongLong, false), - ItemCount((short)0x0014, RntbdTokenType.ULong, false), - RequestCharge((short)0x0015, RntbdTokenType.Double, false), + ReadsPerformed((short) 0x0007, RntbdTokenType.ULong, false), + WritesPerformed((short) 0x0008, RntbdTokenType.ULong, false), + QueriesPerformed((short) 0x0009, RntbdTokenType.ULong, false), + IndexTermsGenerated((short) 0x000A, RntbdTokenType.ULong, false), + ScriptsExecuted((short) 0x000B, RntbdTokenType.ULong, false), + RetryAfterMilliseconds((short) 0x000C, RntbdTokenType.ULong, false), + IndexingDirective((short) 0x000D, RntbdTokenType.Byte, false), + StorageMaxResoureQuota((short) 0x000E, RntbdTokenType.String, false), + StorageResourceQuotaUsage((short) 0x000F, RntbdTokenType.String, false), + SchemaVersion((short) 0x0010, RntbdTokenType.SmallString, false), + CollectionPartitionIndex((short) 0x0011, RntbdTokenType.ULong, false), + CollectionServiceIndex((short) 0x0012, RntbdTokenType.ULong, false), + LSN((short) 0x0013, RntbdTokenType.LongLong, false), + ItemCount((short) 0x0014, RntbdTokenType.ULong, false), + RequestCharge((short) 0x0015, RntbdTokenType.Double, false), // not used((short)0x0016), - OwnerFullName((short)0x0017, RntbdTokenType.String, false), - OwnerId((short)0x0018, RntbdTokenType.String, false), - DatabaseAccountId((short)0x0019, RntbdTokenType.String, false), - QuorumAckedLSN((short)0x001A, RntbdTokenType.LongLong, false), - RequestValidationFailure((short)0x001B, RntbdTokenType.Byte, false), - SubStatus((short)0x001C, RntbdTokenType.ULong, false), - CollectionUpdateProgress((short)0x001D, RntbdTokenType.ULong, false), - CurrentWriteQuorum((short)0x001E, RntbdTokenType.ULong, false), - CurrentReplicaSetSize((short)0x001F, RntbdTokenType.ULong, false), - CollectionLazyIndexProgress((short)0x0020, RntbdTokenType.ULong, false), - PartitionKeyRangeId((short)0x0021, RntbdTokenType.String, false), + OwnerFullName((short) 0x0017, RntbdTokenType.String, false), + OwnerId((short) 0x0018, RntbdTokenType.String, false), + DatabaseAccountId((short) 0x0019, RntbdTokenType.String, false), + QuorumAckedLSN((short) 0x001A, RntbdTokenType.LongLong, false), + RequestValidationFailure((short) 0x001B, RntbdTokenType.Byte, false), + SubStatus((short) 0x001C, RntbdTokenType.ULong, false), + CollectionUpdateProgress((short) 0x001D, RntbdTokenType.ULong, false), + CurrentWriteQuorum((short) 0x001E, RntbdTokenType.ULong, false), + CurrentReplicaSetSize((short) 0x001F, RntbdTokenType.ULong, false), + CollectionLazyIndexProgress((short) 0x0020, RntbdTokenType.ULong, false), + PartitionKeyRangeId((short) 0x0021, RntbdTokenType.String, false), // not used((short)0x0022), RequestMessageReceivedTime // not used((short)0x0023), ResponseMessageSentTime // not used((short)0x0024), ResponseMessageTimeOffset - LogResults((short)0x0025, RntbdTokenType.String, false), - XPRole((short)0x0026, RntbdTokenType.ULong, false), - IsRUPerMinuteUsed((short)0x0027, RntbdTokenType.Byte, false), - QueryMetrics((short)0x0028, RntbdTokenType.String, false), - GlobalCommittedLSN((short)0x0029, RntbdTokenType.LongLong, false), - NumberOfReadRegions((short)0x0030, RntbdTokenType.ULong, false), - OfferReplacePending((short)0x0031, RntbdTokenType.Byte, false), - ItemLSN((short)0x0032, RntbdTokenType.LongLong, false), - RestoreState((short)0x0033, RntbdTokenType.String, false), - CollectionSecurityIdentifier((short)0x0034, RntbdTokenType.String, false), - TransportRequestID((short)0x0035, RntbdTokenType.ULong, false), - ShareThroughput((short)0x0036, RntbdTokenType.Byte, false), + LogResults((short) 0x0025, RntbdTokenType.String, false), + XPRole((short) 0x0026, RntbdTokenType.ULong, false), + IsRUPerMinuteUsed((short) 0x0027, RntbdTokenType.Byte, false), + QueryMetrics((short) 0x0028, RntbdTokenType.String, false), + GlobalCommittedLSN((short) 0x0029, RntbdTokenType.LongLong, false), + NumberOfReadRegions((short) 0x0030, RntbdTokenType.ULong, false), + OfferReplacePending((short) 0x0031, RntbdTokenType.Byte, false), + ItemLSN((short) 0x0032, RntbdTokenType.LongLong, false), + RestoreState((short) 0x0033, RntbdTokenType.String, false), + CollectionSecurityIdentifier((short) 0x0034, RntbdTokenType.String, false), + TransportRequestID((short) 0x0035, RntbdTokenType.ULong, false), + ShareThroughput((short) 0x0036, RntbdTokenType.Byte, false), // not used((short)0x0037), LeaseSeqNumber - DisableRntbdChannel((short)0x0038, RntbdTokenType.Byte, false), - ServerDateTimeUtc((short)0x0039, RntbdTokenType.SmallString, false), - LocalLSN((short)0x003A, RntbdTokenType.LongLong, false), - QuorumAckedLocalLSN((short)0x003B, RntbdTokenType.LongLong, false), - ItemLocalLSN((short)0x003C, RntbdTokenType.LongLong, false), - HasTentativeWrites((short)0x003D, RntbdTokenType.Byte, false), - SessionToken((short)0x003E, RntbdTokenType.String, false); + DisableRntbdChannel((short) 0x0038, RntbdTokenType.Byte, false), + ServerDateTimeUtc((short) 0x0039, RntbdTokenType.SmallString, false), + LocalLSN((short) 0x003A, RntbdTokenType.LongLong, false), + QuorumAckedLocalLSN((short) 0x003B, RntbdTokenType.LongLong, false), + ItemLocalLSN((short) 0x003C, RntbdTokenType.LongLong, false), + HasTentativeWrites((short) 0x003D, RntbdTokenType.Byte, false), + SessionToken((short) 0x003E, RntbdTokenType.String, false); public static final ImmutableMap map; public static final ImmutableSet set = Sets.immutableEnumSet(EnumSet.allOf(RntbdResponseHeader.class)); diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdContext.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdContext.java index 6c3f567769083..38a3e1b3332a8 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdContext.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdContext.java @@ -15,7 +15,7 @@ import java.util.HashMap; import java.util.UUID; -import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.CurrentProtocolVersion; +import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.CURRENT_PROTOCOL_VERSION; import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.RntbdContextHeader; import static com.google.common.base.Preconditions.checkState; @@ -127,14 +127,10 @@ public static RntbdContext decode(final ByteBuf in) { map.put("serverVersion", headers.serverVersion.getValue()); } - headers.releaseBuffers(); throw new RntbdContextException(responseStatus.getStatus(), details, Collections.unmodifiableMap(map)); - - } else { - RntbdContext context = new RntbdContext(responseStatus, headers); - headers.releaseBuffers(); - return context; } + + return new RntbdContext(responseStatus, headers); } public void encode(final ByteBuf out) { @@ -147,7 +143,7 @@ public void encode(final ByteBuf out) { responseStatus.encode(out); headers.encode(out); - headers.releaseBuffers(); + headers.release(); final int end = out.writerIndex(); @@ -165,7 +161,7 @@ public static RntbdContext from(final RntbdContextRequest request, final ServerP headers.clientVersion.setValue(request.getClientVersion()); headers.idleTimeoutInSeconds.setValue(0); - headers.protocolVersion.setValue(CurrentProtocolVersion); + headers.protocolVersion.setValue(CURRENT_PROTOCOL_VERSION); headers.serverAgent.setValue(properties.getAgent()); headers.serverVersion.setValue(properties.getVersion()); headers.unauthenticatedTimeoutInSeconds.setValue(0); diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdContextRequest.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdContextRequest.java index 47d6a14210a5b..182f2ac81c901 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdContextRequest.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdContextRequest.java @@ -16,7 +16,7 @@ import java.util.UUID; import static com.azure.data.cosmos.internal.HttpConstants.Versions; -import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.CurrentProtocolVersion; +import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.CURRENT_PROTOCOL_VERSION; import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.RntbdContextRequestHeader; import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.RntbdOperationType; import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.RntbdResourceType; @@ -118,7 +118,7 @@ private static final class Headers extends RntbdTokenStream { private static final Logger Logger = LoggerFactory.getLogger(RntbdContextRequestEncoder.class); /** - * Returns {@code true} if the given message is an @{link RntbdContextRequest} instance + * Returns {@code true} if the given message is an {@link RntbdContextRequest} message. *

- * If {@code false} this message should be passed to the next @{link ChannelOutboundHandler} in the pipeline. + * If {@code false} this message should be passed to the next {@link io.netty.channel.ChannelHandlerContext} in the + * pipeline. * - * @param message the message to encode - * @return @{code true}, if the given message is an an @{link RntbdContextRequest} instance; otherwise @{false} + * @param message the message to encode. + * + * @return {@code true}, if the given message is an an @{link RntbdContextRequest} instance; {@code false} + * otherwise. */ @Override public boolean acceptOutboundMessage(final Object message) { - return message instanceof RntbdContextRequest; + return message.getClass() == RntbdContextRequest.class; } /** - * Encode an @{link RntbdContextRequest} message into a {@link ByteBuf} + * Encode an {@link RntbdContextRequest} message into a {@link ByteBuf}. *

* This method will be called for each written message that can be handled by this encoder. * - * @param context the {@link ChannelHandlerContext} which this {@link MessageToByteEncoder} belongs to - * @param message the message to encode - * @param out the {@link ByteBuf} into which the encoded message will be written - * @throws IllegalStateException is thrown if an error occurs + * @param context the {@link ChannelHandlerContext} to which this {@link MessageToByteEncoder} belongs. + * @param message the message to encode. + * @param out the {@link ByteBuf} into which the encoded message will be written. + * + * @throws IllegalStateException is thrown if an error occurs. */ @Override - protected void encode(final ChannelHandlerContext context, final Object message, final ByteBuf out) throws IllegalStateException { + protected void encode( + final ChannelHandlerContext context, + final RntbdContextRequest message, + final ByteBuf out) throws IllegalStateException { - final RntbdContextRequest request = (RntbdContextRequest)message; out.markWriterIndex(); try { - request.encode(out); + message.encode(out); } catch (final IllegalStateException error) { out.resetWriterIndex(); throw error; } - Logger.debug("{}: ENCODE COMPLETE: request={}", context.channel(), request); + Logger.debug("{}: ENCODE COMPLETE: message={}", context.channel(), message); } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdMetrics.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdMetrics.java index 090bfc46dfe07..cae62f18d4cbd 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdMetrics.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdMetrics.java @@ -6,23 +6,25 @@ import com.azure.data.cosmos.internal.directconnectivity.RntbdTransportClient; import com.codahale.metrics.ConsoleReporter; import com.codahale.metrics.MetricRegistry; -import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.annotation.JsonPropertyOrder; import com.google.common.net.PercentEscaper; import io.micrometer.core.instrument.Clock; +import io.micrometer.core.instrument.DistributionSummary; import io.micrometer.core.instrument.Gauge; -import io.micrometer.core.instrument.Measurement; import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.Tag; import io.micrometer.core.instrument.Tags; import io.micrometer.core.instrument.Timer; import io.micrometer.core.instrument.composite.CompositeMeterRegistry; import io.micrometer.core.instrument.config.NamingConvention; +import io.micrometer.core.instrument.distribution.HistogramSnapshot; import io.micrometer.core.instrument.dropwizard.DropwizardConfig; import io.micrometer.core.instrument.dropwizard.DropwizardMeterRegistry; import io.micrometer.core.instrument.util.HierarchicalNameMapper; import io.micrometer.core.lang.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import javax.annotation.Nonnull; import java.util.concurrent.TimeUnit; @@ -30,32 +32,38 @@ @SuppressWarnings("UnstableApiUsage") @JsonPropertyOrder({ "tags", "concurrentRequests", "requests", "responseErrors", "responseSuccesses", "completionRate", "responseRate", - "channelsAcquired", "channelsAvailable", "requestQueueLength", "usedDirectMemory", "usedHeapMemory" + "requestSize", "responseSize", "channelsAcquired", "channelsAvailable", "requestQueueLength", "usedDirectMemory", + "usedHeapMemory" }) public final class RntbdMetrics { // region Fields - private static final PercentEscaper escaper = new PercentEscaper("_-", false); + private static final PercentEscaper PERCENT_ESCAPER = new PercentEscaper("_-", false); + + private static final Logger logger = LoggerFactory.getLogger(RntbdMetrics.class); private static final CompositeMeterRegistry registry = new CompositeMeterRegistry(); - private static final String prefix = "azure.cosmos.directTcp."; - private static MeterRegistry consoleLoggingRegistry; + static { + try { + int step = Integer.getInteger("azure.cosmos.monitoring.consoleLogging.step", 0); + if (step > 0) { + RntbdMetrics.add(RntbdMetrics.consoleLoggingRegistry(step)); + } + } catch (Throwable error) { + logger.error("failed to initialize console logging registry due to ", error); + } + } - private final RntbdTransportClient transportClient; private final RntbdEndpoint endpoint; + private final DistributionSummary requestSize; private final Timer requests; private final Timer responseErrors; + private final DistributionSummary responseSize; private final Timer responseSuccesses; private final Tags tags; - - static { - int step = Integer.getInteger("azure.cosmos.monitoring.consoleLogging.step", 0); - if (step > 0) { - RntbdMetrics.add(RntbdMetrics.consoleLoggingRegistry(step)); - } - } + private final RntbdTransportClient transportClient; // endregion @@ -67,6 +75,7 @@ public RntbdMetrics(RntbdTransportClient client, RntbdEndpoint endpoint) { this.endpoint = endpoint; this.tags = Tags.of(client.tag(), endpoint.tag()); + this.requests = registry.timer(nameOf("requests"), tags); this.responseErrors = registry.timer(nameOf("responseErrors"), tags); this.responseSuccesses = registry.timer(nameOf("responseSuccesses"), tags); @@ -102,63 +111,36 @@ public RntbdMetrics(RntbdTransportClient client, RntbdEndpoint endpoint) { .register(registry); Gauge.builder(nameOf("usedDirectMemory"), endpoint, x -> x.usedDirectMemory()) - .description("Java direct memory usage") + .description("Java direct memory usage (MiB)") .baseUnit("bytes") .tags(this.tags) .register(registry); Gauge.builder(nameOf("usedHeapMemory"), endpoint, x -> x.usedHeapMemory()) - .description("Java heap memory usage") + .description("Java heap memory usage (MiB)") .baseUnit("MiB") .tags(this.tags) .register(registry); + + this.requestSize = DistributionSummary.builder(nameOf("requestSize")) + .description("Request size (bytes)") + .baseUnit("bytes") + .tags(this.tags) + .register(registry); + + this.responseSize = DistributionSummary.builder(nameOf("responseSize")) + .description("Response size (bytes)") + .baseUnit("bytes") + .tags(this.tags) + .register(registry); } // endregion // region Accessors - @JsonIgnore - private static synchronized MeterRegistry consoleLoggingRegistry(final int step) { - - if (consoleLoggingRegistry == null) { - - MetricRegistry dropwizardRegistry = new MetricRegistry(); - - ConsoleReporter consoleReporter = ConsoleReporter - .forRegistry(dropwizardRegistry) - .convertRatesTo(TimeUnit.SECONDS) - .convertDurationsTo(TimeUnit.MILLISECONDS) - .build(); - - consoleReporter.start(step, TimeUnit.SECONDS); - - DropwizardConfig dropwizardConfig = new DropwizardConfig() { - - @Override - public String get(@Nullable String key) { - return null; - } - - @Override - public String prefix() { - return "console"; - } - - }; - - consoleLoggingRegistry = new DropwizardMeterRegistry(dropwizardConfig, dropwizardRegistry, HierarchicalNameMapper.DEFAULT, Clock.SYSTEM) { - @Override - @Nonnull - protected Double nullGaugeValue() { - return Double.NaN; - } - }; - - consoleLoggingRegistry.config().namingConvention(NamingConvention.dot); - } - - return consoleLoggingRegistry; + public static void add(MeterRegistry registry) { + RntbdMetrics.registry.add(registry); } @JsonProperty @@ -172,13 +154,13 @@ public int channelsAvailable() { } /*** - * Computes the number of successful (non-error) responses received divided by the number of completed requests + * Computes the number of successful (non-error) responses received divided by the number of completed requests. * - * @return The number of successful (non-error) responses received divided by the number of completed requests + * @return number of successful (non-error) responses received divided by the number of completed requests. */ @JsonProperty public double completionRate() { - return this.responseSuccesses.count() / (double)this.requests.count(); + return this.responseSuccesses.count() / (double) this.requests.count(); } @JsonProperty @@ -197,13 +179,18 @@ public int requestQueueLength() { } @JsonProperty - public Iterable requests() { - return this.requests.measure(); + public HistogramSnapshot requestSize() { + return this.requestSize.takeSnapshot(); + } + + @JsonProperty + public HistogramSnapshot requests() { + return this.requests.takeSnapshot(); } @JsonProperty - public Iterable responseErrors() { - return this.responseErrors.measure(); + public HistogramSnapshot responseErrors() { + return this.responseErrors.takeSnapshot(); } /*** @@ -213,12 +200,17 @@ public Iterable responseErrors() { */ @JsonProperty public double responseRate() { - return this.responseSuccesses.count() / (double)(this.requests.count() + this.endpoint.concurrentRequests()); + return this.responseSuccesses.count() / (double) (this.requests.count() + this.endpoint.concurrentRequests()); } @JsonProperty - public Iterable responseSuccesses() { - return this.responseSuccesses.measure(); + public HistogramSnapshot responseSize() { + return this.responseSize.takeSnapshot(); + } + + @JsonProperty + public HistogramSnapshot responseSuccesses() { + return this.responseSuccesses.takeSnapshot(); } @JsonProperty @@ -240,16 +232,12 @@ public long usedHeapMemory() { // region Methods - public static void add(MeterRegistry registry) { - RntbdMetrics.registry.add(registry); - } - - public void markComplete(RntbdRequestRecord record) { - record.stop(this.requests, record.isCompletedExceptionally() ? this.responseErrors : this.responseSuccesses); - } - - public static String escape(String value) { - return escaper.escape(value); + public void markComplete(RntbdRequestRecord requestRecord) { + requestRecord.stop(this.requests, requestRecord.isCompletedExceptionally() + ? this.responseErrors + : this.responseSuccesses); + this.requestSize.record(requestRecord.requestLength()); + this.responseSize.record(requestRecord.responseLength()); } @Override @@ -261,8 +249,51 @@ public String toString() { // region Private + static String escape(String value) { + return PERCENT_ESCAPER.escape(value); + } + + private static MeterRegistry consoleLoggingRegistry(final int step) { + + final MetricRegistry dropwizardRegistry = new MetricRegistry(); + + ConsoleReporter consoleReporter = ConsoleReporter + .forRegistry(dropwizardRegistry) + .convertRatesTo(TimeUnit.SECONDS) + .convertDurationsTo(TimeUnit.MILLISECONDS) + .build(); + + consoleReporter.start(step, TimeUnit.SECONDS); + + DropwizardConfig dropwizardConfig = new DropwizardConfig() { + + @Override + public String get(@Nullable String key) { + return null; + } + + @Override + public String prefix() { + return "console"; + } + + }; + + final MeterRegistry consoleLoggingRegistry = new DropwizardMeterRegistry( + dropwizardConfig, dropwizardRegistry, HierarchicalNameMapper.DEFAULT, Clock.SYSTEM) { + @Override + @Nonnull + protected Double nullGaugeValue() { + return Double.NaN; + } + }; + + consoleLoggingRegistry.config().namingConvention(NamingConvention.dot); + return consoleLoggingRegistry; + } + private static String nameOf(final String member) { - return prefix + member; + return "azure.cosmos.directTcp." + member; } // endregion diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestArgs.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestArgs.java index 2efa46449ada9..940eed0f55a92 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestArgs.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestArgs.java @@ -15,9 +15,9 @@ import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; -import java.math.BigDecimal; import java.net.URI; import java.time.Duration; +import java.time.OffsetDateTime; import java.util.UUID; import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; @@ -27,7 +27,7 @@ import static io.micrometer.core.instrument.Timer.Sample; @JsonPropertyOrder({ - "transportRequestId", "origin", "replicaPath", "activityId", "operationType", "resourceType", "creationTime", + "transportRequestId", "activityId", "origin", "replicaPath", "operationType", "resourceType", "timeCreated", "lifetime" }) public final class RntbdRequestArgs { @@ -36,7 +36,8 @@ public final class RntbdRequestArgs { private final Sample sample; private final UUID activityId; - private final long creationTime; + private final OffsetDateTime timeCreated; + private final long nanoTimeCreated; private final Stopwatch lifetime; private final String origin; private final URI physicalAddress; @@ -47,7 +48,8 @@ public final class RntbdRequestArgs { public RntbdRequestArgs(final RxDocumentServiceRequest serviceRequest, final URI physicalAddress) { this.sample = Timer.start(); this.activityId = UUID.fromString(serviceRequest.getActivityId()); - this.creationTime = System.nanoTime(); + this.timeCreated = OffsetDateTime.now(); + this.nanoTimeCreated = System.nanoTime(); this.lifetime = Stopwatch.createStarted(); this.origin = physicalAddress.getScheme() + "://" + physicalAddress.getAuthority(); this.physicalAddress = physicalAddress; @@ -63,17 +65,17 @@ public UUID activityId() { return this.activityId; } - @JsonProperty - public long creationTime() { - return this.creationTime; - } - @JsonSerialize(using = ToStringSerializer.class) @JsonProperty public Duration lifetime() { return this.lifetime.elapsed(); } + @JsonIgnore + public long nanoTimeCreated() { + return this.nanoTimeCreated; + } + @JsonProperty public String origin() { return this.origin; @@ -94,6 +96,11 @@ public RxDocumentServiceRequest serviceRequest() { return this.serviceRequest; } + @JsonProperty + public OffsetDateTime timeCreated() { + return this.timeCreated; + } + @JsonProperty public long transportRequestId() { return this.transportRequestId; @@ -114,18 +121,17 @@ public String toString() { return RntbdObjectMapper.toString(this); } - public void traceOperation(final Logger logger, final ChannelHandlerContext context, final String operationName, final Object... args) { + public void traceOperation( + final Logger logger, final ChannelHandlerContext context, final String operationName, final Object... args) { - checkNotNull(logger, "logger"); + checkNotNull(logger, "expected non-null logger"); - if (logger.isTraceEnabled()) { - final BigDecimal lifetime = BigDecimal.valueOf(this.lifetime.elapsed().toNanos(), 6); - logger.trace("{},{},\"{}({})\",\"{}\",\"{}\"", this.creationTime, lifetime, operationName, - Stream.of(args).map(arg -> - arg == null ? "null" : arg.toString()).collect(Collectors.joining(",") - ), - this, context - ); + if (logger.isDebugEnabled()) { + logger.debug("{},{},\"{}({})\",\"{}\",\"{}\"", this.timeCreated, this.lifetime.elapsed(), operationName, + Stream.of(args) + .map(arg -> arg == null ? "null" : arg.toString()) + .collect(Collectors.joining(",")), + this, context); } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestDecoder.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestDecoder.java index 3192bd80c62c7..46cb85f0fb328 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestDecoder.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestDecoder.java @@ -11,7 +11,7 @@ public final class RntbdRequestDecoder extends ByteToMessageDecoder { /** - * Prepare for decoding an @{link RntbdRequest} or fire a channel readTree event to pass the input message along + * Prepare for decoding an @{link RntbdRequest} or fire a channel readTree event to pass the input message along. * * @param context the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to * @param message the message to be decoded @@ -35,18 +35,22 @@ public void channelRead(final ChannelHandlerContext context, final Object messag } /** - * Decode the input {@link ByteBuf} to an RntbdRequest instance + * Decode the input {@link ByteBuf} to an {@link RntbdRequest} instance. *

* This method will be called till either the input {@link ByteBuf} has nothing to readTree after return from this * method or till nothing was readTree from the input {@link ByteBuf}. * - * @param context the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to - * @param in the {@link ByteBuf} from which to readTree data - * @param out the {@link List} to which decoded messages should be added + * @param context the {@link ChannelHandlerContext} to which this {@link ByteToMessageDecoder} belongs. + * @param in the {@link ByteBuf} from which to read data. + * @param out the {@link List} to which decoded messages should be added. + * * @throws IllegalStateException thrown if an error occurs */ @Override - protected void decode(final ChannelHandlerContext context, final ByteBuf in, final List out) throws IllegalStateException { + protected void decode( + final ChannelHandlerContext context, + final ByteBuf in, + final List out) throws IllegalStateException { final RntbdRequest request; in.markReaderIndex(); diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestEncoder.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestEncoder.java index 87e31bbf62274..b8c3ac8fe40ed 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestEncoder.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestEncoder.java @@ -10,21 +10,21 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public final class RntbdRequestEncoder extends MessageToByteEncoder { +public final class RntbdRequestEncoder extends MessageToByteEncoder { private static final Logger logger = LoggerFactory.getLogger(RntbdRequestEncoder.class); /** - * Returns {@code true} if the given message is an @{link RntbdRequest} instance + * Returns {@code true} if the given message is an {@link RntbdRequestRecord} instance. *

* If {@code false} this message should be passed to the next {@link ChannelOutboundHandler} in the pipeline. * * @param message the message to encode - * @return {@code true}, if the given message is an an {@link RntbdRequest} instance; otherwise @{false} + * @return {@code true}, if the given message is an an {@link RntbdRequestRecord} instance; otherwise {@code false}. */ @Override public boolean acceptOutboundMessage(final Object message) { - return message instanceof RntbdRequestArgs; + return message.getClass() == RntbdRequestRecord.class; } /** @@ -34,12 +34,16 @@ public boolean acceptOutboundMessage(final Object message) { * * @param context the {@link ChannelHandlerContext} which this {@link MessageToByteEncoder} belongs encode * @param message the message to encode - * @param out the {@link ByteBuf} into which the encoded message will be written + * @param out the {@link ByteBuf} into which the encoded message will be written */ @Override - protected void encode(final ChannelHandlerContext context, final Object message, final ByteBuf out) throws Exception { + protected void encode( + final ChannelHandlerContext context, + final RntbdRequestRecord message, + final ByteBuf out + ) { - final RntbdRequest request = RntbdRequest.from((RntbdRequestArgs) message); + final RntbdRequest request = RntbdRequest.from(message.args()); final int start = out.writerIndex(); try { @@ -49,9 +53,10 @@ protected void encode(final ChannelHandlerContext context, final Object message, throw error; } + message.requestLength(out.writerIndex() - start); + if (logger.isDebugEnabled()) { - final int length = out.writerIndex() - start; - logger.debug("{}: ENCODE COMPLETE: length={}, request={}", context.channel(), length, request); + logger.debug("{}: ENCODE COMPLETE: request={}", context.channel(), request); } } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestFrame.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestFrame.java index e8dee9ab6a0fa..e5cba2c67de4d 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestFrame.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestFrame.java @@ -164,7 +164,7 @@ private static RntbdOperationType map(final OperationType operationType) { case Stop: return RntbdOperationType.Stop; case SqlQuery: - return RntbdOperationType.SQLQuery; + return RntbdOperationType.SqlQuery; case Update: return RntbdOperationType.Update; case ForceConfigRefresh: diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestHeaders.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestHeaders.java index e3223d17cf054..21f2876adf971 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestHeaders.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestHeaders.java @@ -45,7 +45,7 @@ final class RntbdRequestHeaders extends RntbdTokenStream { // region Fields - private static final String UrlTrim = "/+"; + private static final String URL_TRIM = "/"; // endregion @@ -1004,19 +1004,13 @@ private void addResourceIdOrPathHeaders(final RxDocumentServiceRequest request) // not "apps/appName/partitions/partitionKey/replicas/replicaId/dbs/dbName" final String address = request.getResourceAddress(); - final String[] fragments = address.split(UrlTrim); + final String[] fragments = StringUtils.split(address, URL_TRIM); int count = fragments.length; - int index = 0; - - if (count > 0 && fragments[0].isEmpty()) { - ++index; - --count; - } if (count >= 2) { - switch (fragments[index]) { + switch (fragments[0]) { case Paths.DATABASES_PATH_SEGMENT: - this.getDatabaseName().setValue(fragments[index + 1]); + this.getDatabaseName().setValue(fragments[1]); break; default: final String reason = String.format(Locale.ROOT, RMResources.InvalidResourceAddress, @@ -1026,52 +1020,52 @@ private void addResourceIdOrPathHeaders(final RxDocumentServiceRequest request) } if (count >= 4) { - switch (fragments[index + 2]) { + switch (fragments[2]) { case Paths.COLLECTIONS_PATH_SEGMENT: - this.getCollectionName().setValue(fragments[index + 3]); + this.getCollectionName().setValue(fragments[3]); break; case Paths.USERS_PATH_SEGMENT: - this.getUserName().setValue(fragments[index + 3]); + this.getUserName().setValue(fragments[3]); break; case Paths.USER_DEFINED_TYPES_PATH_SEGMENT: - this.getUserDefinedTypeName().setValue(fragments[index + 3]); + this.getUserDefinedTypeName().setValue(fragments[3]); break; } } if (count >= 6) { - switch (fragments[index + 4]) { + switch (fragments[4]) { case Paths.DOCUMENTS_PATH_SEGMENT: - this.getDocumentName().setValue(fragments[index + 5]); + this.getDocumentName().setValue(fragments[5]); break; case Paths.STORED_PROCEDURES_PATH_SEGMENT: - this.getStoredProcedureName().setValue(fragments[index + 5]); + this.getStoredProcedureName().setValue(fragments[5]); break; case Paths.PERMISSIONS_PATH_SEGMENT: - this.getPermissionName().setValue(fragments[index + 5]); + this.getPermissionName().setValue(fragments[5]); break; case Paths.USER_DEFINED_FUNCTIONS_PATH_SEGMENT: - this.getUserDefinedFunctionName().setValue(fragments[index + 5]); + this.getUserDefinedFunctionName().setValue(fragments[5]); break; case Paths.TRIGGERS_PATH_SEGMENT: - this.getTriggerName().setValue(fragments[index + 5]); + this.getTriggerName().setValue(fragments[5]); break; case Paths.CONFLICTS_PATH_SEGMENT: - this.getConflictName().setValue(fragments[index + 5]); + this.getConflictName().setValue(fragments[5]); break; case Paths.PARTITION_KEY_RANGES_PATH_SEGMENT: - this.getPartitionKeyRangeName().setValue(fragments[index + 5]); + this.getPartitionKeyRangeName().setValue(fragments[5]); break; case Paths.SCHEMAS_PATH_SEGMENT: - this.getSchemaName().setValue(fragments[index + 5]); + this.getSchemaName().setValue(fragments[5]); break; } } if (count >= 8) { - switch (fragments[index + 6]) { + switch (fragments[6]) { case Paths.ATTACHMENTS_PATH_SEGMENT: - this.getAttachmentName().setValue(fragments[index + 7]); + this.getAttachmentName().setValue(fragments[7]); break; } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestManager.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestManager.java index 15add1d09a2dd..533f24c7cc24c 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestManager.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestManager.java @@ -64,9 +64,9 @@ import static com.azure.data.cosmos.internal.HttpConstants.SubStatusCodes; import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdClientChannelHealthChecker.Timestamps; import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.RntbdResponseHeader; -import static com.azure.data.cosmos.internal.guava27.Strings.lenientFormat; import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; +import static com.azure.data.cosmos.internal.guava27.Strings.lenientFormat; public final class RntbdRequestManager implements ChannelHandler, ChannelInboundHandler, ChannelOutboundHandler { @@ -157,11 +157,10 @@ public void channelInactive(final ChannelHandlerContext context) { } /** - * The {@link Channel} of the {@link ChannelHandlerContext} has read a message from its peer - *

+ * The {@link Channel} of the {@link ChannelHandlerContext} has read a message from its peer. * - * @param context {@link ChannelHandlerContext} to which this {@link RntbdRequestManager} belongs - * @param message The message read + * @param context {@link ChannelHandlerContext} to which this {@link RntbdRequestManager} belongs. + * @param message The message read. */ @Override public void channelRead(final ChannelHandlerContext context, final Object message) { @@ -169,7 +168,7 @@ public void channelRead(final ChannelHandlerContext context, final Object messag this.traceOperation(context, "channelRead"); try { - if (message instanceof RntbdResponse) { + if (message.getClass() == RntbdResponse.class) { try { this.messageReceived(context, (RntbdResponse) message); @@ -200,7 +199,7 @@ public void channelRead(final ChannelHandlerContext context, final Object messag } /** - * The {@link Channel} of the {@link ChannelHandlerContext} has fully consumed the most-recent message read + * The {@link Channel} of the {@link ChannelHandlerContext} has fully consumed the most-recent message read. *

* If {@link ChannelOption#AUTO_READ} is off, no further attempt to read inbound data from the current * {@link Channel} will be made until {@link ChannelHandlerContext#read} is called. This leaves time @@ -483,17 +482,15 @@ public void write(final ChannelHandlerContext context, final Object message, fin this.traceOperation(context, "write", message); - if (message instanceof RntbdRequestRecord) { + if (message.getClass() == RntbdRequestRecord.class) { - RntbdRequestRecord record = (RntbdRequestRecord) message; + final RntbdRequestRecord record = (RntbdRequestRecord) message; this.timestamps.channelWriteAttempted(); - context.writeAndFlush(this.addPendingRequestRecord(context, record), promise).addListener(completed -> { + context.write(this.addPendingRequestRecord(context, record), promise).addListener(completed -> { + record.stage(RntbdRequestRecord.Stage.SENT); if (completed.isSuccess()) { - record.stage(RntbdRequestRecord.Stage.SENT); this.timestamps.channelWriteCompleted(); - } else { - record.stage(RntbdRequestRecord.Stage.UNSENT); } }); @@ -502,7 +499,7 @@ public void write(final ChannelHandlerContext context, final Object message, fin if (message == RntbdHealthCheckRequest.MESSAGE) { - context.writeAndFlush(RntbdHealthCheckRequest.MESSAGE, promise).addListener(completed -> { + context.write(RntbdHealthCheckRequest.MESSAGE, promise).addListener(completed -> { if (completed.isSuccess()) { this.timestamps.channelPingCompleted(); } @@ -511,7 +508,10 @@ public void write(final ChannelHandlerContext context, final Object message, fin return; } - final IllegalStateException error = new IllegalStateException(lenientFormat("message of %s: %s", message.getClass(), message)); + final IllegalStateException error = new IllegalStateException(lenientFormat("message of %s: %s", + message.getClass(), + message)); + reportIssue(context, "", error); this.exceptionCaught(context, error); } @@ -557,12 +557,11 @@ Timestamps snapshotTimestamps() { // region Private methods - private RntbdRequestArgs addPendingRequestRecord(final ChannelHandlerContext context, final RntbdRequestRecord record) { + private RntbdRequestRecord addPendingRequestRecord(final ChannelHandlerContext context, final RntbdRequestRecord record) { return this.pendingRequests.compute(record.transportRequestId(), (id, current) -> { reportIssueUnless(current == null, context, "id: {}, current: {}, request: {}", record); - record.stage(RntbdRequestRecord.Stage.QUEUED); final Timeout pendingRequestTimeout = record.newTimeout(timeout -> { @@ -581,9 +580,9 @@ private RntbdRequestArgs addPendingRequestRecord(final ChannelHandlerContext con pendingRequestTimeout.cancel(); }); - return record.stage(RntbdRequestRecord.Stage.QUEUED); + return record; - }).args(); + }); } private void completeAllPendingRequestsExceptionally( @@ -678,10 +677,10 @@ private void completeAllPendingRequestsExceptionally( } /** - * This method is called for each incoming message of type {@link StoreResponse} to complete a request + * This method is called for each incoming message of type {@link RntbdResponse} to complete a request. * - * @param context {@link ChannelHandlerContext} encode to which this {@link RntbdRequestManager} belongs - * @param response the message encode handle + * @param context {@link ChannelHandlerContext} to which this {@link RntbdRequestManager request manager} belongs. + * @param response the {@link RntbdResponse message} received. */ private void messageReceived(final ChannelHandlerContext context, final RntbdResponse response) { @@ -699,10 +698,14 @@ private void messageReceived(final ChannelHandlerContext context, final RntbdRes return; } + requestRecord.responseLength(response.getMessageLength()); + requestRecord.stage(RntbdRequestRecord.Stage.RECEIVED); + final HttpResponseStatus status = response.getStatus(); final UUID activityId = response.getActivityId(); + final int statusCode = status.code(); - if (HttpResponseStatus.OK.code() <= status.code() && status.code() < HttpResponseStatus.MULTIPLE_CHOICES.code()) { + if (HttpResponseStatus.OK.code() <= statusCode && statusCode < HttpResponseStatus.MULTIPLE_CHOICES.code()) { final StoreResponse storeResponse = response.toStoreResponse(this.contextFuture.getNow(null)); requestRecord.complete(storeResponse); @@ -720,9 +723,9 @@ private void messageReceived(final ChannelHandlerContext context, final RntbdRes // ..Create Error instance - final CosmosError error = response.hasPayload() ? - BridgeInternal.createCosmosError(RntbdObjectMapper.readTree(response)) : - new CosmosError(Integer.toString(status.code()), status.reasonPhrase(), status.codeClass().name()); + final CosmosError error = response.hasPayload() + ? BridgeInternal.createCosmosError(RntbdObjectMapper.readTree(response)) + : new CosmosError(Integer.toString(statusCode), status.reasonPhrase(), status.codeClass().name()); // ..Map RNTBD response headers to HTTP response headers @@ -845,7 +848,7 @@ private static void reportIssueUnless( } private void traceOperation(final ChannelHandlerContext context, final String operationName, final Object... args) { - logger.trace("{}\n{}\n{}", operationName, context, args); + logger.debug("{}\n{}\n{}", operationName, context, args); } // endregion diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestRecord.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestRecord.java index c952152cc7280..081c1bb740a62 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestRecord.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdRequestRecord.java @@ -5,8 +5,8 @@ import com.azure.data.cosmos.BridgeInternal; import com.azure.data.cosmos.RequestTimeoutException; +import com.azure.data.cosmos.internal.RequestTimeline; import com.azure.data.cosmos.internal.directconnectivity.StoreResponse; -import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.databind.SerializerProvider; import com.fasterxml.jackson.databind.annotation.JsonSerialize; @@ -14,34 +14,61 @@ import io.micrometer.core.instrument.Timer; import io.netty.util.Timeout; import io.netty.util.TimerTask; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.time.Duration; +import java.time.OffsetDateTime; import java.util.UUID; import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +import static com.azure.data.cosmos.internal.guava27.Strings.lenientFormat; import static com.google.common.base.Preconditions.checkNotNull; @JsonSerialize(using = RntbdRequestRecord.JsonSerializer.class) public final class RntbdRequestRecord extends CompletableFuture { - private static final AtomicReferenceFieldUpdater - stageUpdater = AtomicReferenceFieldUpdater.newUpdater(RntbdRequestRecord.class, Stage.class, "stage"); + private static final Logger logger = LoggerFactory.getLogger(RntbdRequestRecord.class); + + private static final AtomicIntegerFieldUpdater REQUEST_LENGTH = + AtomicIntegerFieldUpdater.newUpdater(RntbdRequestRecord.class, "requestLength"); + + private static final AtomicIntegerFieldUpdater RESPONSE_LENGTH = + AtomicIntegerFieldUpdater.newUpdater(RntbdRequestRecord.class, "responseLength"); + + private static final AtomicReferenceFieldUpdater STAGE = + AtomicReferenceFieldUpdater.newUpdater( + RntbdRequestRecord.class, + Stage.class, + "stage"); private final RntbdRequestArgs args; private final RntbdRequestTimer timer; + + private volatile int requestLength; + private volatile int responseLength; private volatile Stage stage; + private volatile OffsetDateTime timeCompleted; + private volatile OffsetDateTime timePipelined; + private volatile OffsetDateTime timeQueued; + private volatile OffsetDateTime timeSent; + private volatile OffsetDateTime timeReceived; + public RntbdRequestRecord(final RntbdRequestArgs args, final RntbdRequestTimer timer) { - checkNotNull(args, "args"); - checkNotNull(timer, "timer"); + checkNotNull(args, "expected non-null args"); + checkNotNull(timer, "expected non-null timer"); - this.stage = Stage.CREATED; + this.timeQueued = OffsetDateTime.now(); + this.requestLength = -1; + this.responseLength = -1; + this.stage = Stage.QUEUED; this.args = args; this.timer = timer; } @@ -56,35 +83,101 @@ public RntbdRequestArgs args() { return this.args; } - public long creationTime() { - return this.args.creationTime(); + public Duration lifetime() { + return this.args.lifetime(); } - public boolean expire() { - final RequestTimeoutException error = new RequestTimeoutException(this.toString(), this.args.physicalAddress()); - BridgeInternal.setRequestHeaders(error, this.args.serviceRequest().getHeaders()); - return this.completeExceptionally(error); + public int requestLength() { + return this.requestLength; } - public Duration lifetime() { - return this.args.lifetime(); + RntbdRequestRecord requestLength(int value) { + REQUEST_LENGTH.set(this, value); + return this; } - public Timeout newTimeout(final TimerTask task) { - return this.timer.newTimeout(task); + public int responseLength() { + return this.responseLength; + } + + RntbdRequestRecord responseLength(int value) { + RESPONSE_LENGTH.set(this, value); + return this; } public Stage stage() { - return stageUpdater.get(this); + return this.stage; } - public RntbdRequestRecord stage(Stage value) { - stageUpdater.set(this, value); + public RntbdRequestRecord stage(final Stage value) { + + final OffsetDateTime time = OffsetDateTime.now(); + + STAGE.updateAndGet(this, current -> { + + switch (value) { + case PIPELINED: + if (current != Stage.QUEUED) { + logger.debug("Expected transition from QUEUED to PIPELINED, not {} to PIPELINED", current); + break; + } + this.timePipelined = time; + break; + case SENT: + if (current != Stage.PIPELINED) { + logger.debug("Expected transition from PIPELINED to SENT, not {} to SENT", current); + break; + } + this.timeSent = time; + break; + case RECEIVED: + if (current != Stage.SENT) { + logger.debug("Expected transition from SENT to RECEIVED, not {} to RECEIVED", current); + break; + } + this.timeReceived = time; + break; + case COMPLETED: + if (current == Stage.COMPLETED) { + logger.debug("Request already COMPLETED", current); + break; + } + this.timeCompleted = time; + break; + default: + throw new IllegalStateException(lenientFormat("there is no transition from %s to %s", + current, + value)); + } + + return value; + }); + return this; } - public long timeoutIntervalInMillis() { - return this.timer.getRequestTimeout(TimeUnit.MILLISECONDS); + public OffsetDateTime timeCompleted() { + return this.timeCompleted; + } + + public OffsetDateTime timeCreated() { + return this.args.timeCreated(); + } + + public OffsetDateTime timePipelined() { + return this.timePipelined; + } + + public OffsetDateTime timeQueued() { + return this.timeQueued; + } + + public OffsetDateTime timeReceived() { + return this.timeReceived; + } + + public OffsetDateTime timeSent() { + return this.timeSent; } public long transportRequestId() { @@ -95,6 +188,43 @@ public long transportRequestId() { // region Methods + public boolean expire() { + final RequestTimeoutException error = new RequestTimeoutException(this.toString(), this.args.physicalAddress()); + BridgeInternal.setRequestHeaders(error, this.args.serviceRequest().getHeaders()); + return this.completeExceptionally(error); + } + + public Timeout newTimeout(final TimerTask task) { + return this.timer.newTimeout(task); + } + + public RequestTimeline takeTimelineSnapshot() { + + OffsetDateTime now = OffsetDateTime.now(); + + OffsetDateTime timeCreated = this.timeCreated(); + OffsetDateTime timeQueued = this.timeQueued(); + OffsetDateTime timePipelined = this.timePipelined(); + OffsetDateTime timeSent = this.timeSent(); + OffsetDateTime timeReceived = this.timeReceived(); + OffsetDateTime timeCompleted = this.timeCompleted(); + OffsetDateTime timeCompletedOrNow = timeCompleted == null ? now : timeCompleted; + + return RequestTimeline.of( + new RequestTimeline.Event("created", + timeCreated, timeQueued == null ? timeCompletedOrNow : timeQueued), + new RequestTimeline.Event("queued", + timeQueued, timePipelined == null ? timeCompletedOrNow : timePipelined), + new RequestTimeline.Event("pipelined", + timePipelined, timeSent == null ? timeCompletedOrNow : timeSent), + new RequestTimeline.Event("transitTime", + timeSent, timeReceived == null ? timeCompletedOrNow : timeReceived), + new RequestTimeline.Event("received", + timeReceived, timeCompletedOrNow), + new RequestTimeline.Event("completed", + timeCompleted, now)); + } + public long stop(Timer requests, Timer responses) { return this.args.stop(requests, responses); } @@ -109,11 +239,13 @@ public String toString() { // region Types public enum Stage { - CREATED, QUEUED, SENT, UNSENT, CANCELLED_BY_CLIENT + QUEUED, PIPELINED, SENT, RECEIVED, COMPLETED } static final class JsonSerializer extends StdSerializer { + private static final long serialVersionUID = -6869331366500298083L; + JsonSerializer() { super(RntbdRequestRecord.class); } @@ -125,6 +257,12 @@ public void serialize( final SerializerProvider provider) throws IOException { generator.writeStartObject(); + generator.writeObjectField("args", value.args()); + generator.writeNumberField("requestLength", value.requestLength()); + generator.writeNumberField("responseLength", value.responseLength()); + + // status + generator.writeObjectFieldStart("status"); generator.writeBooleanField("done", value.isDone()); generator.writeBooleanField("cancelled", value.isCancelled()); @@ -155,7 +293,8 @@ public void serialize( } generator.writeEndObject(); - generator.writeObjectField("args", value.args); + + generator.writeObjectField("timeline", value.takeTimelineSnapshot()); generator.writeEndObject(); } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdResponse.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdResponse.java index d1b8fdb31673f..110a6c54c33ef 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdResponse.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdResponse.java @@ -15,6 +15,7 @@ import io.netty.buffer.ByteBufUtil; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.util.IllegalReferenceCountException; import io.netty.util.ReferenceCounted; import io.netty.util.ResourceLeakDetector; @@ -22,20 +23,23 @@ import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.UUID; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.RntbdResponseHeader; +import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.base.Preconditions.checkState; -import static java.lang.Math.min; +import static java.lang.Integer.min; -@JsonPropertyOrder({ "frame", "headers", "content" }) +@JsonPropertyOrder({ "messageLength", "referenceCount", "frame", "headers", "content" }) public final class RntbdResponse implements ReferenceCounted { // region Fields + private static final AtomicIntegerFieldUpdater REFERENCE_COUNT = + AtomicIntegerFieldUpdater.newUpdater(RntbdResponse.class, "referenceCount"); + @JsonSerialize(using = PayloadSerializer.class) - @JsonProperty private final ByteBuf content; @JsonProperty @@ -44,33 +48,64 @@ public final class RntbdResponse implements ReferenceCounted { @JsonProperty private final RntbdResponseHeaders headers; - private final ByteBuf in; + private final ByteBuf message; + + @JsonProperty + private final int messageLength; - private final AtomicInteger referenceCount = new AtomicInteger(); + @JsonProperty + private volatile int referenceCount; // endregion - public RntbdResponse(final UUID activityId, final int statusCode, final Map map, final ByteBuf content) { + // region Constructors + + /** + * Initializes a new {@link RntbdResponse} instance. + *

+ * This method is provided for testing purposes only. It should not be used in product code. + * + * @param activityId an activity ID + * @param statusCode a response status code. + * @param map a collection of response headers. + * @param content a body to be copied to the response. + */ + public RntbdResponse( + final UUID activityId, + final int statusCode, + final Map map, + final ByteBuf content) { this.headers = RntbdResponseHeaders.fromMap(map, content.readableBytes() > 0); - this.in = Unpooled.EMPTY_BUFFER; - this.content = content.copy().retain(); + this.message = Unpooled.EMPTY_BUFFER; + this.content = content.copy(); final HttpResponseStatus status = HttpResponseStatus.valueOf(statusCode); final int length = RntbdResponseStatus.LENGTH + this.headers.computeLength(); this.frame = new RntbdResponseStatus(length, status, activityId); + this.messageLength = length + this.content.writerIndex(); + this.referenceCount = 0; } private RntbdResponse( - final ByteBuf in, final RntbdResponseStatus frame, final RntbdResponseHeaders headers, final ByteBuf content - ) { - this.in = in.retain(); + final ByteBuf message, + final RntbdResponseStatus frame, + final RntbdResponseHeaders headers, + final ByteBuf content) { + + this.message = message; + this.referenceCount = 0; this.frame = frame; this.headers = headers; - this.content = content.retain(); + this.content = content; + this.messageLength = message.writerIndex();; } + // endregion + + // region Accessors + @JsonIgnore public UUID getActivityId() { return this.frame.getActivityId(); @@ -86,6 +121,11 @@ public RntbdResponseHeaders getHeaders() { return this.headers; } + @JsonIgnore + public int getMessageLength() { + return this.messageLength; + } + @JsonIgnore public HttpResponseStatus getStatus() { return this.frame.getStatus(); @@ -96,36 +136,15 @@ public Long getTransportRequestId() { return this.getHeader(RntbdResponseHeader.TransportRequestID); } - static RntbdResponse decode(final ByteBuf in) { - - final int start = in.markReaderIndex().readerIndex(); - - final RntbdResponseStatus frame = RntbdResponseStatus.decode(in); - final RntbdResponseHeaders headers = RntbdResponseHeaders.decode(in.readSlice(frame.getHeadersLength())); - final boolean hasPayload = headers.isPayloadPresent(); - final ByteBuf content; - - if (hasPayload) { - - if (!RntbdFramer.canDecodePayload(in)) { - headers.releaseBuffers(); - in.resetReaderIndex(); - return null; - } - - content = in.readSlice(in.readIntLE()); - - } else { - - content = Unpooled.EMPTY_BUFFER; - } - - final int end = in.readerIndex(); - in.resetReaderIndex(); + // endregion - return new RntbdResponse(in.readSlice(end - start), frame, headers, content); - } + // region Methods + /** + * Serializes the current {@link RntbdResponse response} to the given {@link ByteBuf byte buffer}. + * + * @param out the output {@link ByteBuf byte buffer}. + */ public void encode(final ByteBuf out) { final int start = out.writerIndex(); @@ -144,12 +163,24 @@ public void encode(final ByteBuf out) { } } - @JsonIgnore + /** + * Returns the value of the given {@link RntbdResponse response} {@link RntbdResponseHeader header}. + * + * @param header the {@link RntbdResponse response} {@link RntbdResponseHeader header}. + * @param the {@link RntbdResponse response} {@link RntbdResponseHeader header} value type. + * + * @return the value of the given {@code header}. + */ @SuppressWarnings("unchecked") public T getHeader(final RntbdResponseHeader header) { - return (T)this.headers.get(header).getValue(); + return (T) this.headers.get(header).getValue(); } + /** + * Returns {@code true} if this {@link RntbdResponse response} has a payload. + * + * @return {@code true} if this {@link RntbdResponse response} has a payload; {@code false} otherwise. + */ public boolean hasPayload() { return this.headers.isPayloadPresent(); } @@ -159,11 +190,13 @@ public boolean hasPayload() { */ @Override public int refCnt() { - return this.referenceCount.get(); + return this.referenceCount; } /** - * Decreases the reference count by {@code 1} and deallocate this response if the count reaches {@code 0}. + * Decreases the reference count by {@code 1}. + *

+ * The current {@link RntbdResponse response} is deallocated if the count reaches {@code 0}. * * @return {@code true} if and only if the reference count became {@code 0} and this response is deallocated. */ @@ -176,35 +209,29 @@ public boolean release() { * Decreases the reference count by {@code decrement} and deallocates this response if the count reaches {@code 0}. * * @param decrement amount of the decrease. + * * @return {@code true} if and only if the reference count became {@code 0} and this response has been deallocated. */ @Override public boolean release(final int decrement) { - return this.referenceCount.accumulateAndGet(decrement, (value, n) -> { - - value = value - min(value, n); + checkArgument(decrement > 0, "expected decrement, not %s", decrement); - if (value == 0) { + return REFERENCE_COUNT.accumulateAndGet(this, decrement, (referenceCount, decrease) -> { - checkState(this.headers != null && this.content != null); - this.headers.releaseBuffers(); + if (referenceCount < decrement) { + throw new IllegalReferenceCountException(referenceCount, -decrease); + }; - if (this.in != Unpooled.EMPTY_BUFFER) { - this.in.release(); - } + referenceCount = referenceCount - decrease; - if (this.content != Unpooled.EMPTY_BUFFER) { + if (referenceCount == 0) { this.content.release(); + this.headers.release(); + this.message.release(); } - // TODO: DANOBLE: figure out why PooledUnsafeDirectByteBuf violates these expectations: - // checkState(this.in == Unpooled.EMPTY_BUFFER || this.in.refCnt() == 0); - // checkState(this.content == Unpooled.EMPTY_BUFFER || this.content.refCnt() == 0); - // Specifically, why are this.in.refCnt() and this.content.refCnt() equal to 1? - } - - return value; + return referenceCount; }) == 0; } @@ -213,9 +240,8 @@ public boolean release(final int decrement) { * Increases the reference count by {@code 1}. */ @Override - public ReferenceCounted retain() { - this.referenceCount.incrementAndGet(); - return this; + public RntbdResponse retain() { + return this.retain(1); } /** @@ -224,21 +250,20 @@ public ReferenceCounted retain() { * @param increment amount of the increase */ @Override - public ReferenceCounted retain(final int increment) { - this.referenceCount.addAndGet(increment); - return this; - } + public RntbdResponse retain(final int increment) { - StoreResponse toStoreResponse(final RntbdContext context) { + checkArgument(increment > 0, "expected positive increment, not %s", increment); - checkNotNull(context, "context"); - final int length = this.content.readableBytes(); + REFERENCE_COUNT.accumulateAndGet(this, increment, (referenceCount, increase) -> { + if (referenceCount == 0) { + this.content.retain(); + this.headers.retain(); + this.message.retain(); + } + return referenceCount + increase; + }); - return new StoreResponse( - this.getStatus().code(), - this.headers.asList(context, this.getActivityId()), - length == 0 ? null : this.content.readCharSequence(length, StandardCharsets.UTF_8).toString() - ); + return this; } @Override @@ -249,35 +274,83 @@ public String toString() { /** * Records the current access location of this object for debugging purposes *

- * If this object is determined to be leaked, the information recorded by this operation will be provided to you - * via {@link ResourceLeakDetector}. This method is a shortcut to {@link #touch(Object) touch(null)}. + * If this object is determined to be leaked, the information recorded by this operation will be provided to you via + * {@link ResourceLeakDetector}. This method is a shortcut to {@link #touch(Object) touch(null)}. */ @Override - public ReferenceCounted touch() { + public RntbdResponse touch() { return this; } /** * Records the current access location of this object with additional arbitrary information for debugging purposes *

- * If this object is determined to be leaked, the information recorded by this operation will be - * provided to you via {@link ResourceLeakDetector}. + * If this object is determined to be leaked, the information recorded by this operation will be provided to you via + * {@link ResourceLeakDetector}. * * @param hint information useful for debugging (unused) */ @Override - public ReferenceCounted touch(final Object hint) { + public RntbdResponse touch(final Object hint) { return this; } + static RntbdResponse decode(final ByteBuf in) { + + final int start = in.markReaderIndex().readerIndex(); + + final RntbdResponseStatus frame = RntbdResponseStatus.decode(in); + final RntbdResponseHeaders headers = RntbdResponseHeaders.decode(in.readSlice(frame.getHeadersLength())); + final boolean hasPayload = headers.isPayloadPresent(); + final ByteBuf content; + + if (hasPayload) { + + if (!RntbdFramer.canDecodePayload(in)) { + in.resetReaderIndex(); + return null; + } + + content = in.readSlice(in.readIntLE()); + + } else { + + content = Unpooled.EMPTY_BUFFER; + } + + final int end = in.readerIndex(); + in.resetReaderIndex(); + + return new RntbdResponse(in.readSlice(end - start), frame, headers, content); + } + + StoreResponse toStoreResponse(final RntbdContext context) { + + checkNotNull(context, "expected non-null context"); + + final int contentLength = this.content.writerIndex(); + + return new StoreResponse( + this.getStatus().code(), + this.headers.asList(context, this.getActivityId()), + contentLength == 0 ? null : this.content.getCharSequence(0, contentLength, StandardCharsets.UTF_8).toString()); + } + + // endregion + + // region Types + private static class PayloadSerializer extends StdSerializer { - public PayloadSerializer() { + PayloadSerializer() { super(ByteBuf.class, true); } @Override - public void serialize(final ByteBuf value, final JsonGenerator generator, final SerializerProvider provider) throws IOException { + public void serialize( + final ByteBuf value, + final JsonGenerator generator, + final SerializerProvider provider) throws IOException { final int length = value.readableBytes(); @@ -288,4 +361,6 @@ public void serialize(final ByteBuf value, final JsonGenerator generator, final generator.writeEndObject(); } } + + // endregion } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdResponseDecoder.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdResponseDecoder.java index 1c963e97cfd49..a4a02585a10a0 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdResponseDecoder.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdResponseDecoder.java @@ -13,16 +13,16 @@ public final class RntbdResponseDecoder extends ByteToMessageDecoder { - private static final Logger Logger = LoggerFactory.getLogger(RntbdResponseDecoder.class); + private static final Logger logger = LoggerFactory.getLogger(RntbdResponseDecoder.class); /** - * Deserialize from an input {@link ByteBuf} to an {@link RntbdResponse} instance + * Deserialize from an input {@link ByteBuf} to an {@link RntbdResponse} instance. *

- * This method is called till it reads no bytes from the {@link ByteBuf} or there is no more data to be readTree. + * This method is called till it reads no bytes from the {@link ByteBuf} or there is no more data to be read. * - * @param context the {@link ChannelHandlerContext} to which this {@link RntbdResponseDecoder} belongs - * @param in the {@link ByteBuf} to which data to be decoded is readTree - * @param out the {@link List} to which decoded messages are added + * @param context the {@link ChannelHandlerContext} to which this {@link RntbdResponseDecoder} belongs. + * @param in the {@link ByteBuf} to which data to be decoded is read. + * @param out the {@link List} to which decoded messages are added. */ @Override protected void decode(final ChannelHandlerContext context, final ByteBuf in, final List out) { @@ -32,10 +32,9 @@ protected void decode(final ChannelHandlerContext context, final ByteBuf in, fin final RntbdResponse response = RntbdResponse.decode(in); if (response != null) { - Logger.debug("{} DECODE COMPLETE: {}", context.channel(), response); + logger.debug("{} DECODE COMPLETE: {}", context.channel(), response); in.discardReadBytes(); - response.retain(); - out.add(response); + out.add(response.retain()); } } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdServiceEndpoint.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdServiceEndpoint.java index e37cb29641f16..accfed34f56f4 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdServiceEndpoint.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdServiceEndpoint.java @@ -16,7 +16,6 @@ import io.netty.bootstrap.Bootstrap; import io.netty.channel.AdaptiveRecvByteBufAllocator; import io.netty.channel.Channel; -import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelOption; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; @@ -173,7 +172,7 @@ public RntbdRequestRecord request(final RntbdRequestArgs args) { this.throwIfClosed(); this.concurrentRequests.incrementAndGet(); - this.lastRequestTime.set(args.creationTime()); + this.lastRequestTime.set(args.nanoTimeCreated()); if (logger.isDebugEnabled()) { args.traceOperation(logger, null, "request"); @@ -239,11 +238,7 @@ private RntbdRequestRecord write(final RntbdRequestArgs requestArgs) { requestArgs.traceOperation(logger, null, "write"); final Channel channel = (Channel)connected.get(); this.releaseToPool(channel); - - channel.write(requestRecord).addListener((ChannelFuture future) -> { - requestArgs.traceOperation(logger, null, "writeComplete", channel); - }); - + channel.write(requestRecord.stage(RntbdRequestRecord.Stage.PIPELINED)); return; } @@ -320,10 +315,8 @@ public Provider(final RntbdTransportClient transportClient, final Options option final int threadCount = Runtime.getRuntime().availableProcessors(); final LogLevel wireLogLevel; - if (logger.isTraceEnabled()) { + if (logger.isDebugEnabled()) { wireLogLevel = LogLevel.TRACE; - } else if (logger.isDebugEnabled()) { - wireLogLevel = LogLevel.DEBUG; } else { wireLogLevel = null; } @@ -346,8 +339,6 @@ public void close() { if (this.closed.compareAndSet(false, true)) { - this.requestTimer.close(); - for (final RntbdEndpoint endpoint : this.endpoints.values()) { endpoint.close(); } @@ -361,6 +352,7 @@ public void close() { logger.error("\n [{}]\n failed to close endpoints due to ", this, future.cause()); }); + this.requestTimer.close(); return; } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdToken.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdToken.java index b7f2c9b1eed2e..11650ec9f46b7 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdToken.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdToken.java @@ -14,10 +14,10 @@ import io.netty.handler.codec.CorruptedFrameException; import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.RntbdHeader; +import static com.azure.data.cosmos.internal.guava27.Strings.lenientFormat; import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.base.Preconditions.checkState; -import static com.azure.data.cosmos.internal.guava27.Strings.lenientFormat; @JsonPropertyOrder({ "id", "name", "type", "present", "required", "value" }) final class RntbdToken { @@ -74,15 +74,13 @@ public Object getValue() { } if (this.value instanceof ByteBuf) { - final ByteBuf buffer = (ByteBuf)this.value; - this.value = codec.defaultValue(); + final ByteBuf buffer = (ByteBuf) this.value; try { + this.value = codec.defaultValue(); this.value = codec.read(buffer); } catch (final CorruptedFrameException error) { String message = lenientFormat("failed to read %s value: %s", this.getName(), error.getMessage()); throw new CorruptedFrameException(message); - } finally { - buffer.release(); } } else { this.value = codec.convert(this.value); @@ -98,7 +96,6 @@ public T getValue(final Class cls) { @JsonProperty public void setValue(final Object value) { this.ensureValid(value); - this.releaseBuffer(); this.value = value; this.length = Integer.MIN_VALUE; } @@ -129,7 +126,7 @@ public int computeLength() { } if (this.value instanceof ByteBuf) { - final ByteBuf buffer = (ByteBuf)this.value; + final ByteBuf buffer = (ByteBuf) this.value; checkState(buffer.readerIndex() == 0); return HEADER_LENGTH + buffer.readableBytes(); } @@ -148,12 +145,7 @@ public static RntbdToken create(final RntbdHeader header) { public void decode(final ByteBuf in) { checkNotNull(in, "expected non-null in"); - - if (this.value instanceof ByteBuf) { - ((ByteBuf)this.value).release(); - } - - this.value = this.header.type().codec().readSlice(in).retain(); // No data transfer until first call to RntbdToken.getValue + this.value = this.header.type().codec().readSlice(in); } public void encode(final ByteBuf out) { @@ -172,17 +164,13 @@ public void encode(final ByteBuf out) { out.writeByte(this.getTokenType().id()); if (this.value instanceof ByteBuf) { - out.writeBytes((ByteBuf)this.value); + out.writeBytes((ByteBuf) this.value); } else { this.ensureValid(this.value); this.header.type().codec().write(this.value, out); } } - public boolean releaseBuffer() { - return this.value instanceof ByteBuf && ((ByteBuf)this.value).release(); - } - @Override public String toString() { return RntbdObjectMapper.toString(this); @@ -193,8 +181,10 @@ public String toString() { // region Privates private void ensureValid(final Object value) { - checkArgument(value != null, "value: null"); - checkArgument(this.header.type().codec().isValid(value), "value: %s = %s", value.getClass().getName(), value); + checkArgument(value != null, "expected non-null value"); + checkArgument(this.header.type().codec().isValid(value), "invalid value: %s = %s", + value.getClass().getName(), + value); } // endregion @@ -204,13 +194,17 @@ private void ensureValid(final Object value) { static class PropertyFilter extends SimpleBeanPropertyFilter { @Override - public void serializeAsField(final Object object, final JsonGenerator generator, final SerializerProvider provider, final PropertyWriter writer) throws Exception { + public void serializeAsField( + final Object object, + final JsonGenerator generator, + final SerializerProvider provider, + final PropertyWriter writer) throws Exception { if (generator.canOmitFields()) { final Object value = writer.getMember().getValue(object); - if (value instanceof RntbdToken && !((RntbdToken)value).isPresent()) { + if (value instanceof RntbdToken && !((RntbdToken) value).isPresent()) { return; } } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdTokenStream.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdTokenStream.java index 1438fec92d9c1..6c572a0178c30 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdTokenStream.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdTokenStream.java @@ -8,15 +8,16 @@ import com.google.common.collect.Maps; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.CorruptedFrameException; +import io.netty.util.ReferenceCounted; import java.util.stream.Collector; import static com.azure.data.cosmos.internal.directconnectivity.rntbd.RntbdConstants.RntbdHeader; -import static com.google.common.base.Preconditions.checkNotNull; import static com.azure.data.cosmos.internal.guava27.Strings.lenientFormat; +import static com.google.common.base.Preconditions.checkNotNull; @SuppressWarnings("UnstableApiUsage") -abstract class RntbdTokenStream & RntbdHeader> { +abstract class RntbdTokenStream & RntbdHeader> implements ReferenceCounted { final ByteBuf in; final ImmutableMap headers; @@ -24,16 +25,18 @@ abstract class RntbdTokenStream & RntbdHeader> { RntbdTokenStream(final ImmutableSet headers, final ImmutableMap ids, final ByteBuf in) { - checkNotNull(headers, "headers"); - checkNotNull(ids, "ids"); - checkNotNull(in, "in"); + checkNotNull(headers, "expected non-null headers"); + checkNotNull(ids, "expected non-null ids"); + checkNotNull(in, "expected non-null in"); final Collector> collector = Maps.toImmutableEnumMap(h -> h, RntbdToken::create); this.tokens = headers.stream().collect(collector); this.headers = ids; - this.in = in.retain(); + this.in = in; } + // region Methods + final int computeCount() { int count = 0; @@ -96,13 +99,46 @@ final RntbdToken get(final T header) { return this.tokens.get(header); } - final void releaseBuffers() { - for (final RntbdToken token : this.tokens.values()) { - token.releaseBuffer(); + @Override + public final int refCnt() { + return this.in.refCnt(); + } + + @Override + public final boolean release() { + return this.release(1); + } + + @Override + public final boolean release(final int count) { + return this.in.release(count); + } + + @Override + public final RntbdTokenStream retain() { + return this.retain(1); + } + + @Override + public final RntbdTokenStream retain(final int count) { + this.in.retain(count); + return this; } - in.release(); + + @Override + public ReferenceCounted touch(Object hint) { + return this; } + @Override + public ReferenceCounted touch() { + return this; + } + + // endregion + + // region Types + private static final class UndefinedHeader implements RntbdHeader { private final short id; @@ -133,4 +169,6 @@ public RntbdTokenType type() { return this.type; } } + + // endregion } diff --git a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdTokenType.java b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdTokenType.java index bf6f325fb0eb6..96f41e8e34be0 100644 --- a/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdTokenType.java +++ b/sdk/cosmos/microsoft-azure-cosmos/src/main/java/com/azure/data/cosmos/internal/directconnectivity/rntbd/RntbdTokenType.java @@ -11,8 +11,8 @@ import java.nio.charset.StandardCharsets; import java.util.UUID; -import static com.google.common.base.Preconditions.checkState; import static com.azure.data.cosmos.internal.guava27.Strings.lenientFormat; +import static com.google.common.base.Preconditions.checkState; enum RntbdTokenType {