Skip to content

Add support for specifying default Direct TCP options using System properties #299

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kushagraThapar merged 8 commits into from
Dec 18, 2019
Merged

Add support for specifying default Direct TCP options using System properties #299

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
775a113
Configs.getProtocol now accepts in priority order -Dcosmos.directMode…
Dec 15, 2019
4410763
ReadMyWritesConsistencyTest now sets document data field size and Rnt…
Dec 16, 2019
2847990
Removed option to specify document data field size on ReadMyWritesCon…
Dec 16, 2019
4db2feb
Enclosed static initializer in try/finally to reduce chance of unexpe…
Dec 17, 2019
716eced
Bug fix and tweaks based on code review feedback
Dec 17, 2019
d9f86ff
Increased request timer resolution from 5 to 100 ms
Dec 17, 2019
b7bb71d
Decreased request timer resolution from 100 to 5 ms
Dec 17, 2019
c24ac96
Added RntbdTransportClient.Options.requestTimerResolution property fo…
Dec 17, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion benchmark/src/main/java/com/microsoft/azure/cosmosdb/benchmark/ReadMyWriteWorkflow.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@
import com.microsoft.azure.cosmosdb.SqlQuerySpec;
import com.microsoft.azure.cosmosdb.internal.Utils;
import com.microsoft.azure.cosmosdb.rx.internal.NotFoundException;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import rx.Observable;
import rx.Subscriber;
Expand Down
9 changes: 5 additions & 4 deletions ...ark/src/test/java/com/microsoft/azure/cosmosdb/benchmark/ReadMyWritesConsistencyTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,10 +69,10 @@ public class ReadMyWritesConsistencyTest {
StringUtils.defaultString(Strings.emptyToNull(
System.getenv().get("DESIRED_CONSISTENCY")), "Session"));

private final String directModeProtocol =
System.getProperty("azure.cosmos.directModeProtocol",
private final String documentDataFieldSize =
System.getProperty("DOCUMENT_DATA_FIELD_SIZE",
StringUtils.defaultString(Strings.emptyToNull(
System.getenv().get("DIRECT_MODE_PROTOCOL")), Protocol.Tcp.name()));
System.getenv().get("DOCUMENT_DATA_FIELD_SIZE")), "20"));

private final int initialCollectionThroughput = 10_000;

Expand Down Expand Up @@ -105,6 +105,7 @@ public void readMyWrites(boolean useNameLink) throws Exception {
" -operation ReadMyWrites" +
" -connectionMode Direct" +
" -numberOfPreCreatedDocuments 100" +
" -documentDataFieldSize %s" +
" -printingInterval 60" +
"%s";

Expand All @@ -117,12 +118,12 @@ public void readMyWrites(boolean useNameLink) throws Exception {
concurrency,
numberOfOperationsAsString,
maxRunningTime,
documentDataFieldSize,
(useNameLink ? " -useNameLink" : ""));

Configuration cfg = new Configuration();
new JCommander(cfg, StringUtils.split(cmd));

logger.info("azure.cosmos.directModeProtocol={}, {}", directModeProtocol, cfg);
AtomicInteger success = new AtomicInteger();
AtomicInteger error = new AtomicInteger();

Expand Down
8 changes: 7 additions & 1 deletion commons/src/main/java/com/microsoft/azure/cosmosdb/rx/internal/Configs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,13 @@ public SslContext getSslContext() {
}

public Protocol getProtocol() {
String protocol = getJVMConfigAsString(PROTOCOL, DEFAULT_PROTOCOL.name());

String protocol = getJVMConfigAsString(PROTOCOL, StringUtils.defaultString(
StringUtils.defaultString(
System.getProperty("azure.cosmos.directModeProtocol"),
System.getenv("DIRECT_MODE_PROTOCOL")),
DEFAULT_PROTOCOL.name()));

try {
return Protocol.valueOf(WordUtils.capitalize(protocol.toLowerCase()));
} catch (Exception e) {
Expand Down
6 changes: 6 additions & 0 deletions direct-impl/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ SOFTWARE.
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<test.groups>unit</test.groups>
<guava.version>27.0.1-jre</guava.version>
<jackson-datatype.version>2.9.9</jackson-datatype.version>
Comment thread
David-Noble-at-work marked this conversation as resolved.
Outdated
<metrics.version>4.1.0</metrics.version>
<micrometer.version>1.2.0</micrometer.version>
</properties>
Expand Down Expand Up @@ -295,6 +296,11 @@ SOFTWARE.
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
<version>${jackson-datatype.version}</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
Expand Down
185 changes: 136 additions & 49 deletions ...n/java/com/microsoft/azure/cosmosdb/internal/directconnectivity/RntbdTransportClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@

package com.microsoft.azure.cosmosdb.internal.directconnectivity;

import com.fasterxml.jackson.annotation.JsonIgnore;
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;
Expand All @@ -45,7 +47,9 @@
import rx.Single;
import rx.SingleEmitter;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.time.Duration;
import java.util.Iterator;
Expand Down Expand Up @@ -190,36 +194,73 @@ public static final class Options {

// region Fields

@JsonProperty()
private final int bufferPageSize;
private final String certificateHostNameOverride;

@JsonProperty()
private final Duration connectionTimeout;

@JsonProperty()
private final Duration idleChannelTimeout;

@JsonProperty()
private final Duration idleEndpointTimeout;

@JsonProperty()
private final int maxBufferCapacity;

@JsonProperty()
private final int maxChannelsPerEndpoint;

@JsonProperty()
private final int maxRequestsPerChannel;
private final int partitionCount;

@JsonProperty()
private final Duration receiveHangDetectionTime;

@JsonProperty()
private final Duration requestExpiryInterval;

@JsonProperty()
private final Duration requestTimeout;

@JsonProperty()
private final Duration sendHangDetectionTime;

@JsonProperty()
private final Duration shutdownTimeout;

@JsonIgnore()
private final UserAgentContainer userAgent;

// endregion

// region Constructors

private Options() {
this.bufferPageSize = 8192;
this.connectionTimeout = null;
this.idleChannelTimeout = Duration.ZERO;
this.idleEndpointTimeout = Duration.ofSeconds(70L);
this.maxBufferCapacity = 8192 << 10;
this.maxChannelsPerEndpoint = 10;
this.maxRequestsPerChannel = 30;
this.receiveHangDetectionTime = Duration.ofSeconds(65L);
this.requestExpiryInterval = Duration.ofSeconds(5L);
this.requestTimeout = null;
this.sendHangDetectionTime = Duration.ofSeconds(10L);
this.shutdownTimeout = Duration.ofSeconds(15L);
this.userAgent = new UserAgentContainer();
}

private Options(Builder builder) {
this.bufferPageSize = builder.bufferPageSize;
this.certificateHostNameOverride = builder.certificateHostNameOverride;
this.connectionTimeout = builder.connectionTimeout == null ? builder.requestTimeout : builder.connectionTimeout;
this.idleChannelTimeout = builder.idleChannelTimeout;
this.idleEndpointTimeout = builder.idleEndpointTimeout;
this.maxBufferCapacity = builder.maxBufferCapacity;
this.maxChannelsPerEndpoint = builder.maxChannelsPerEndpoint;
this.maxRequestsPerChannel = builder.maxRequestsPerChannel;
this.partitionCount = builder.partitionCount;
this.receiveHangDetectionTime = builder.receiveHangDetectionTime;
this.requestExpiryInterval = builder.requestExpiryInterval;
this.requestTimeout = builder.requestTimeout;
Expand All @@ -236,10 +277,6 @@ public int bufferPageSize() {
return this.bufferPageSize;
}

public String certificateHostNameOverride() {
return this.certificateHostNameOverride;
}

public Duration connectionTimeout() {
return this.connectionTimeout;
}
Expand All @@ -264,10 +301,6 @@ public int maxRequestsPerChannel() {
return this.maxRequestsPerChannel;
}

public int partitionCount() {
return this.partitionCount;
}

public Duration receiveHangDetectionTime() {
return this.receiveHangDetectionTime;
}
Expand All @@ -288,17 +321,17 @@ public Duration shutdownTimeout() {
return this.shutdownTimeout;
}

@Override
public String toString() {
return RntbdObjectMapper.toJson(this);
public UserAgentContainer userAgent() {
return this.userAgent;
}

// endregion

// region Methods

public UserAgentContainer userAgent() {
return this.userAgent;
@Override
public String toString() {
return RntbdObjectMapper.toJson(this);
}

// endregion
Expand All @@ -310,35 +343,100 @@ public static class Builder {

// region Fields

private static final UserAgentContainer DEFAULT_USER_AGENT_CONTAINER = new UserAgentContainer();
private static final Duration FIFTEEN_SECONDS = Duration.ofSeconds(15L);
private static final Duration FIVE_SECONDS =Duration.ofSeconds(5L);
private static final Duration SEVENTY_SECONDS = Duration.ofSeconds(70L);
private static final Duration SIXTY_FIVE_SECONDS = Duration.ofSeconds(65L);
private static final Duration TEN_SECONDS = Duration.ofSeconds(10L);

private int bufferPageSize = 8192;
private String certificateHostNameOverride = null;
private Duration connectionTimeout = null;
private Duration idleChannelTimeout = Duration.ZERO;
private Duration idleEndpointTimeout = SEVENTY_SECONDS;
private int maxBufferCapacity = 8192 << 10;
private int maxChannelsPerEndpoint = 10;
private int maxRequestsPerChannel = 30;
private int partitionCount = 1;
private Duration receiveHangDetectionTime = SIXTY_FIVE_SECONDS;
private Duration requestExpiryInterval = FIVE_SECONDS;
private static final Options DEFAULT_OPTIONS;

static {
Comment thread
David-Noble-at-work marked this conversation as resolved.

// 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.

String propertyName = "azure.cosmos.directTcp.defaultOptions";
String string = System.getProperty(propertyName);
Options options = null;

if (string != null) {
// Attempt to set default Direct TCP options based on the JSON string value of "{propertyName}"
try {
options = RntbdObjectMapper.readValue(string, Options.class);
} catch (IOException error) {
logger.error("failed to parse default Direct TCP options {} due to ", string, error);
}
}

if (options == null) {

String path = System.getProperty(propertyName + "File");

if (path != null) {
// Attempt to load default Direct TCP options from the JSON file on the path specified by
// "{propertyName}File"
try {
options = RntbdObjectMapper.readValue(new File(path), Options.class);
} catch (IOException error) {
logger.error("failed to load default Direct TCP options from {} due to ", path, error);
}
}
}

if (options == null) {

String name = propertyName + ".json";
InputStream stream = RntbdTransportClient.class.getClassLoader().getResourceAsStream(name);

if (stream != null) {
// Attempt to load default Direct TCP options from the JSON resource file "{propertyName}.json"
try {
options = RntbdObjectMapper.readValue(stream, Options.class);
} catch (IOException error) {
logger.error("failed to load Direct TCP options from resource {} due to ", name, error);
}
}
}

DEFAULT_OPTIONS = options != null ? options : new Options();
}

private int bufferPageSize;
private Duration connectionTimeout;
private Duration idleChannelTimeout;
private Duration idleEndpointTimeout;
private int maxBufferCapacity;
private int maxChannelsPerEndpoint;
private int maxRequestsPerChannel;
private Duration receiveHangDetectionTime;
private Duration requestExpiryInterval;
private Duration requestTimeout;
private Duration sendHangDetectionTime = TEN_SECONDS;
private Duration shutdownTimeout = FIFTEEN_SECONDS;
private UserAgentContainer userAgent = DEFAULT_USER_AGENT_CONTAINER;
private Duration sendHangDetectionTime;
private Duration shutdownTimeout;
private UserAgentContainer userAgent;

// endregion

// region Constructors

public Builder(Duration requestTimeout) {

this.requestTimeout(requestTimeout);

this.bufferPageSize = DEFAULT_OPTIONS.bufferPageSize;
this.connectionTimeout = DEFAULT_OPTIONS.connectionTimeout;
this.idleChannelTimeout = DEFAULT_OPTIONS.idleChannelTimeout;
this.idleEndpointTimeout = DEFAULT_OPTIONS.idleEndpointTimeout;
this.maxBufferCapacity = DEFAULT_OPTIONS.maxBufferCapacity;
this.maxChannelsPerEndpoint = DEFAULT_OPTIONS.maxChannelsPerEndpoint;
this.maxRequestsPerChannel = DEFAULT_OPTIONS.maxRequestsPerChannel;
this.receiveHangDetectionTime = DEFAULT_OPTIONS.receiveHangDetectionTime;
this.requestExpiryInterval = DEFAULT_OPTIONS.requestExpiryInterval;
this.sendHangDetectionTime = DEFAULT_OPTIONS.sendHangDetectionTime;
this.shutdownTimeout = DEFAULT_OPTIONS.shutdownTimeout;
this.userAgent = DEFAULT_OPTIONS.userAgent;
}

public Builder(int requestTimeoutInSeconds) {
Expand All @@ -365,11 +463,6 @@ public Options build() {
return new Options(this);
}

public Builder certificateHostNameOverride(final String value) {
this.certificateHostNameOverride = value;
return this;
}

public Builder connectionTimeout(final Duration value) {
checkArgument(value == null || value.compareTo(Duration.ZERO) > 0,
"expected positive value, not %s",
Expand Down Expand Up @@ -412,12 +505,6 @@ public Builder maxRequestsPerChannel(final int value) {
return this;
}

public Builder partitionCount(final int value) {
checkArgument(value > 0, "expected positive value, not %s", value);
this.partitionCount = value;
return this;
}

public Builder receiveHangDetectionTime(final Duration value) {
checkArgument(value != null && value.compareTo(Duration.ZERO) > 0,
"expected positive value, not %s",
Expand Down
Loading