Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
## 1.5.0 [unreleased]

### Features

1. [#289](https://github.com/InfluxCommunity/influxdb3-java/pull/289) Add the possibility to disable gRPC compression via the `disableGRPCCompression` parameter in the `ClientConfig`.

### CI

1. [#283](https://github.com/InfluxCommunity/influxdb3-java/pull/283) Fix pipeline not downloading the correct java images.
Expand Down
39 changes: 37 additions & 2 deletions src/main/java/com/influxdb/v3/client/config/ClientConfig.java
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
* <li><code>authenticator</code> - HTTP proxy authenticator</li>
* <li><code>headers</code> - headers to be added to requests</li>
* <li><code>sslRootsFilePath</code> - path to the stored certificates file in PEM format</li>
* <li><code>disableGRPCCompression</code> - disables the default gRPC compression header</li>
* </ul>
* <p>
* If you want to create a client with custom configuration, you can use following code:
Expand Down Expand Up @@ -113,6 +114,7 @@ public final class ClientConfig {
private final Authenticator authenticator;
private final Map<String, String> headers;
private final String sslRootsFilePath;
private final boolean disableGRPCCompression;
Comment thread
vlastahajek marked this conversation as resolved.

/**
* Deprecated use {@link #proxyUrl}.
Expand Down Expand Up @@ -318,6 +320,15 @@ public Map<String, String> getHeaders() {
return headers;
}

/**
* Is gRPC compression disabled.
*
* @return true if gRPC compression is disabled
*/
public boolean getDisableGRPCCompression() {
return disableGRPCCompression;
}

/**
* Validates the configuration properties.
*/
Expand Down Expand Up @@ -354,7 +365,8 @@ public boolean equals(final Object o) {
&& Objects.equals(proxyUrl, that.proxyUrl)
&& Objects.equals(authenticator, that.authenticator)
&& Objects.equals(headers, that.headers)
&& Objects.equals(sslRootsFilePath, that.sslRootsFilePath);
&& Objects.equals(sslRootsFilePath, that.sslRootsFilePath)
&& disableGRPCCompression == that.disableGRPCCompression;
}

@Override
Expand All @@ -363,7 +375,7 @@ public int hashCode() {
database, writePrecision, gzipThreshold, writeNoSync,
timeout, writeTimeout, queryTimeout, allowHttpRedirects, disableServerCertificateValidation,
proxy, proxyUrl, authenticator, headers,
defaultTags, sslRootsFilePath);
defaultTags, sslRootsFilePath, disableGRPCCompression);
}

@Override
Expand All @@ -386,6 +398,7 @@ public String toString() {
.add("headers=" + headers)
.add("defaultTags=" + defaultTags)
.add("sslRootsFilePath=" + sslRootsFilePath)
.add("disableGRPCCompression=" + disableGRPCCompression)
.toString();
}

Expand Down Expand Up @@ -415,6 +428,7 @@ public static final class Builder {
private Authenticator authenticator;
private Map<String, String> headers;
private String sslRootsFilePath;
private boolean disableGRPCCompression;

/**
* Sets the URL of the InfluxDB server.
Expand Down Expand Up @@ -697,6 +711,18 @@ public Builder sslRootsFilePath(@Nullable final String sslRootsFilePath) {
return this;
}

/**
* Sets whether to disable gRPC compression. Default is 'false'.
*
* @param disableGRPCCompression disable gRPC compression
* @return this
*/
@Nonnull
public Builder disableGRPCCompression(final boolean disableGRPCCompression) {
this.disableGRPCCompression = disableGRPCCompression;
return this;
}

/**
* Build an instance of {@code ClientConfig}.
*
Expand Down Expand Up @@ -745,6 +771,9 @@ public ClientConfig build(@Nonnull final String connectionString) throws Malform
if (parameters.containsKey("writeNoSync")) {
this.writeNoSync(Boolean.parseBoolean(parameters.get("writeNoSync")));
}
if (parameters.containsKey("disableGRPCCompression")) {
this.disableGRPCCompression(Boolean.parseBoolean(parameters.get("disableGRPCCompression")));
}

return new ClientConfig(this);
}
Expand Down Expand Up @@ -807,6 +836,11 @@ public ClientConfig build(@Nonnull final Map<String, String> env, final Properti
long to = Long.parseLong(queryTimeout);
this.queryTimeout(Duration.ofSeconds(to));
}
final String disableGRPCCompression = get.apply("INFLUX_DISABLE_GRPC_COMPRESSION",
"influx.disableGRPCCompression");
if (disableGRPCCompression != null) {
this.disableGRPCCompression(Boolean.parseBoolean(disableGRPCCompression));
}

return new ClientConfig(this);
}
Expand Down Expand Up @@ -862,5 +896,6 @@ private ClientConfig(@Nonnull final Builder builder) {
authenticator = builder.authenticator;
headers = builder.headers;
sslRootsFilePath = builder.sslRootsFilePath;
disableGRPCCompression = builder.disableGRPCCompression;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.grpc.Codec;
import io.grpc.DecompressorRegistry;
import io.grpc.HttpConnectProxiedSocketAddress;
import io.grpc.Metadata;
import io.grpc.ProxyDetector;
Expand Down Expand Up @@ -93,8 +95,6 @@ final class FlightSqlClient implements AutoCloseable {
defaultHeaders.put("Authorization", "Bearer " + new String(config.getToken()));
}

defaultHeaders.put("User-Agent", Identity.getUserAgent());

if (config.getHeaders() != null) {
defaultHeaders.putAll(config.getHeaders());
}
Expand Down Expand Up @@ -148,6 +148,8 @@ private FlightClient createFlightClient(@Nonnull final ClientConfig config) {
URI uri = createLocation(config).getUri();
final NettyChannelBuilder nettyChannelBuilder = NettyChannelBuilder.forAddress(uri.getHost(), uri.getPort());

nettyChannelBuilder.userAgent(Identity.getUserAgent());

if (LocationSchemes.GRPC_TLS.equals(uri.getScheme())) {
nettyChannelBuilder.useTransportSecurity();

Expand All @@ -169,6 +171,11 @@ private FlightClient createFlightClient(@Nonnull final ClientConfig config) {
nettyChannelBuilder.maxTraceEvents(0)
.maxInboundMetadataSize(Integer.MAX_VALUE);

if (config.getDisableGRPCCompression()) {
nettyChannelBuilder.decompressorRegistry(DecompressorRegistry.emptyInstance()
.with(Codec.Identity.NONE, false));
}

return FlightGrpcUtils.createFlightClient(new RootAllocator(Long.MAX_VALUE), nettyChannelBuilder.build());
}

Expand Down
3 changes: 3 additions & 0 deletions src/test/java/com/influxdb/v3/client/ITQueryWrite.java
Original file line number Diff line number Diff line change
Expand Up @@ -439,6 +439,9 @@ public void queryTimeoutSuperceededByGrpcOptTest() {
Assertions.assertThat(thrown.getMessage()).matches(".*deadline.*exceeded.*");
}

@EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_URL", matches = ".*")
@EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_TOKEN", matches = ".*")
@EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_DATABASE", matches = ".*")
@Test
public void repeatQueryWithTimeoutTest() {
long timeout = 1000;
Expand Down
92 changes: 92 additions & 0 deletions src/test/java/com/influxdb/v3/client/TestUtils.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
/*
* The MIT License
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package com.influxdb.v3.client;

import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import javax.annotation.Nonnull;

import org.apache.arrow.flight.FlightServer;
import org.apache.arrow.flight.Location;
import org.apache.arrow.flight.NoOpFlightProducer;
import org.apache.arrow.flight.Ticket;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VarCharVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.types.pojo.ArrowType;
import org.apache.arrow.vector.types.pojo.Field;
import org.apache.arrow.vector.types.pojo.FieldType;
import org.apache.arrow.vector.types.pojo.Schema;

public final class TestUtils {

private TestUtils() {
throw new IllegalStateException("Utility class");
}

public static FlightServer simpleFlightServer(@Nonnull final URI uri,
@Nonnull final BufferAllocator allocator,
@Nonnull final NoOpFlightProducer producer) throws Exception {
Location location = Location.forGrpcInsecure(uri.getHost(), uri.getPort());
return FlightServer.builder(allocator, location, producer).build();
}

public static NoOpFlightProducer simpleProducer(@Nonnull final VectorSchemaRoot vectorSchemaRoot) {
return new NoOpFlightProducer() {
@Override
public void getStream(final CallContext context,
final Ticket ticket,
final ServerStreamListener listener) {
listener.start(vectorSchemaRoot);
if (listener.isReady()) {
listener.putNext();
}
listener.completed();
}
};
}

public static VectorSchemaRoot generateVectorSchemaRoot(final int fieldCount, final int rowCount) {
List<Field> fields = new ArrayList<>();
for (int i = 0; i < fieldCount; i++) {
Field field = new Field("field" + i, FieldType.nullable(new ArrowType.Utf8()), null);
fields.add(field);
}

Schema schema = new Schema(fields);
VectorSchemaRoot vectorSchemaRoot = VectorSchemaRoot.create(schema, new RootAllocator(Long.MAX_VALUE));
for (Field field : fields) {
VarCharVector vector = (VarCharVector) vectorSchemaRoot.getVector(field);
vector.allocateNew(rowCount);
for (int i = 0; i < rowCount; i++) {
vector.set(i, "Value".getBytes(StandardCharsets.UTF_8));
}
}
vectorSchemaRoot.setRowCount(rowCount);

return vectorSchemaRoot;
}
}

14 changes: 11 additions & 3 deletions src/test/java/com/influxdb/v3/client/config/ClientConfigTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,8 @@ class ClientConfigTest {
.queryTimeout(Duration.ofSeconds(120))
.allowHttpRedirects(true)
.disableServerCertificateValidation(true)
.headers(Map.of("X-device", "ab-01"));
.headers(Map.of("X-device", "ab-01"))
.disableGRPCCompression(true);

@Test
void equalConfig() {
Expand Down Expand Up @@ -81,6 +82,7 @@ void toStringConfig() {
Assertions.assertThat(configString).contains("timeout=PT30S");
Assertions.assertThat(configString).contains("writeTimeout=PT35S");
Assertions.assertThat(configString).contains("queryTimeout=PT2M");
Assertions.assertThat(configString).contains("disableGRPCCompression=true");

}

Expand Down Expand Up @@ -131,10 +133,11 @@ void fromConnectionString() throws MalformedURLException {

cfg = new ClientConfig.Builder()
.build("http://localhost:9999/"
+ "?token=my-token&authScheme=my-auth");
+ "?token=my-token&authScheme=my-auth&disableGRPCCompression=true");
Assertions.assertThat(cfg.getHost()).isEqualTo("http://localhost:9999/");
Assertions.assertThat(cfg.getToken()).isEqualTo("my-token".toCharArray());
Assertions.assertThat(cfg.getAuthScheme()).isEqualTo("my-auth");
Assertions.assertThat(cfg.getDisableGRPCCompression()).isEqualTo(true);
}

@Test
Expand Down Expand Up @@ -204,7 +207,9 @@ void fromEnv() {
"INFLUX_DATABASE", "my-db",
"INFLUX_PRECISION", "ms",
"INFLUX_GZIP_THRESHOLD", "64",
"INFLUX_WRITE_NO_SYNC", "true"
"INFLUX_WRITE_NO_SYNC", "true",
"INFLUX_DISABLE_GRPC_COMPRESSION", "true"

);
cfg = new ClientConfig.Builder()
.build(env, null);
Expand All @@ -215,6 +220,7 @@ void fromEnv() {
Assertions.assertThat(cfg.getWritePrecision()).isEqualTo(WritePrecision.MS);
Assertions.assertThat(cfg.getGzipThreshold()).isEqualTo(64);
Assertions.assertThat(cfg.getWriteNoSync()).isEqualTo(true);
Assertions.assertThat(cfg.getDisableGRPCCompression()).isTrue();
}

@Test
Expand Down Expand Up @@ -318,6 +324,7 @@ void fromSystemProperties() {
properties.put("influx.precision", "ms");
properties.put("influx.gzipThreshold", "64");
properties.put("influx.writeNoSync", "true");
properties.put("influx.disableGRPCCompression", "true");
cfg = new ClientConfig.Builder()
.build(new HashMap<>(), properties);
Assertions.assertThat(cfg.getHost()).isEqualTo("http://localhost:9999/");
Expand All @@ -327,6 +334,7 @@ void fromSystemProperties() {
Assertions.assertThat(cfg.getWritePrecision()).isEqualTo(WritePrecision.MS);
Assertions.assertThat(cfg.getGzipThreshold()).isEqualTo(64);
Assertions.assertThat(cfg.getWriteNoSync()).isEqualTo(true);
Assertions.assertThat(cfg.getDisableGRPCCompression()).isTrue();
}

@Test
Expand Down
Loading
Loading