From 2bb86a42ad41c613cc14f824ff4b42d341cedcde Mon Sep 17 00:00:00 2001 From: dyma solovei Date: Mon, 26 Oct 2026 11:00:28 +0100 Subject: [PATCH 1/2] feat(grpc): provide an option to configure gRPC channel directly This will allow users to set parameters specific to their use-case, which might not be broadly applicable to others. --- .../io/weaviate/client6/v1/api/Config.java | 31 +++++++-- .../internal/grpc/DefaultGrpcTransport.java | 14 ++-- .../v1/internal/grpc/GrpcChannelOptions.java | 66 ++++++++++++++++--- 3 files changed, 93 insertions(+), 18 deletions(-) diff --git a/src/main/java/io/weaviate/client6/v1/api/Config.java b/src/main/java/io/weaviate/client6/v1/api/Config.java index c3c16623e..dff690add 100644 --- a/src/main/java/io/weaviate/client6/v1/api/Config.java +++ b/src/main/java/io/weaviate/client6/v1/api/Config.java @@ -1,5 +1,7 @@ package io.weaviate.client6.v1.api; +import static java.util.Objects.requireNonNull; + import java.net.URI; import java.util.HashMap; import java.util.Map; @@ -7,6 +9,7 @@ import javax.net.ssl.TrustManagerFactory; +import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; import io.weaviate.client6.v1.internal.BuildInfo; import io.weaviate.client6.v1.internal.ObjectBuilder; import io.weaviate.client6.v1.internal.Proxy; @@ -26,7 +29,8 @@ public record Config( Authentication authentication, TrustManagerFactory trustManagerFactory, Timeout timeout, - Proxy proxy) { + Proxy proxy, + Function channelBuilder) { public static Config of(Function> fn) { return fn.apply(new Custom()).build(); @@ -43,7 +47,8 @@ private Config(Builder builder) { builder.authentication, builder.trustManagerFactory, builder.timeout, - builder.proxy); + builder.proxy, + builder.channelBuilder); } RestTransportOptions restTransportOptions() { @@ -51,7 +56,8 @@ RestTransportOptions restTransportOptions() { } RestTransportOptions restTransportOptions(TokenProvider tokenProvider) { - return new RestTransportOptions(scheme, httpHost, httpPort, headers, tokenProvider, trustManagerFactory, timeout, proxy); + return new RestTransportOptions(scheme, httpHost, httpPort, headers, tokenProvider, trustManagerFactory, timeout, + proxy); } GrpcChannelOptions grpcTransportOptions() { @@ -59,7 +65,8 @@ GrpcChannelOptions grpcTransportOptions() { } GrpcChannelOptions grpcTransportOptions(TokenProvider tokenProvider) { - return new GrpcChannelOptions(scheme, grpcHost, grpcPort, headers, tokenProvider, trustManagerFactory, timeout, proxy); + return new GrpcChannelOptions(scheme, grpcHost, grpcPort, headers, tokenProvider, trustManagerFactory, timeout, + proxy, channelBuilder); } private abstract static class Builder> implements ObjectBuilder { @@ -74,6 +81,7 @@ private abstract static class Builder> implements O protected Timeout timeout = new Timeout(); protected Map headers = new HashMap<>(); protected Proxy proxy; + protected Function channelBuilder; /** * Set URL scheme. Subclasses may increase the visibility of this method to @@ -188,6 +196,21 @@ public SelfT proxy(Proxy proxy) { return (SelfT) this; } + /** + * Override or extend any of configurations previously set for the gRPC channel. + * + *
+ * This is best used to access the parameters specific to gRPC, which the client + * does not expose directly, e.g. keepalive configuration. For other + * configurations, like timeout and proxy, prefer using the APIs exposed by this + * class. + */ + @SuppressWarnings("unchecked") + public SelfT grpcChannelOptions(Function channelBuilder) { + this.channelBuilder = requireNonNull(channelBuilder, "channelBuilder"); + return (SelfT) this; + } + /** * Weaviate will use the URL in this header to call Weaviate Embeddings * Service if an appropriate vectorizer is configured for collection. diff --git a/src/main/java/io/weaviate/client6/v1/internal/grpc/DefaultGrpcTransport.java b/src/main/java/io/weaviate/client6/v1/internal/grpc/DefaultGrpcTransport.java index 7ae6ba8f6..6002c31a6 100644 --- a/src/main/java/io/weaviate/client6/v1/internal/grpc/DefaultGrpcTransport.java +++ b/src/main/java/io/weaviate/client6/v1/internal/grpc/DefaultGrpcTransport.java @@ -2,6 +2,8 @@ import static java.util.Objects.requireNonNull; +import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.util.OptionalInt; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executor; @@ -12,6 +14,7 @@ import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; + import io.grpc.HttpConnectProxiedSocketAddress; import io.grpc.ManagedChannel; import io.grpc.StatusRuntimeException; @@ -29,12 +32,6 @@ import io.weaviate.client6.v1.internal.grpc.protocol.WeaviateProtoBatch.BatchStreamReply; import io.weaviate.client6.v1.internal.grpc.protocol.WeaviateProtoBatch.BatchStreamRequest; -import javax.net.ssl.SSLException; -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; - public final class DefaultGrpcTransport implements GrpcTransport { /** * ListenableFuture callbacks are executed @@ -175,6 +172,11 @@ private static ManagedChannel buildChannel(GrpcChannelOptions transportOptions) } channel.intercept(MetadataUtils.newAttachHeadersInterceptor(transportOptions.headers())); + + if (transportOptions.channelBuilder() != null) { + channel = transportOptions.channelBuilder().apply(channel); + } + return channel.build(); } diff --git a/src/main/java/io/weaviate/client6/v1/internal/grpc/GrpcChannelOptions.java b/src/main/java/io/weaviate/client6/v1/internal/grpc/GrpcChannelOptions.java index 69553a858..57a421007 100644 --- a/src/main/java/io/weaviate/client6/v1/internal/grpc/GrpcChannelOptions.java +++ b/src/main/java/io/weaviate/client6/v1/internal/grpc/GrpcChannelOptions.java @@ -2,10 +2,12 @@ import java.util.Map; import java.util.OptionalInt; +import java.util.function.Function; import javax.net.ssl.TrustManagerFactory; import io.grpc.Metadata; +import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; import io.weaviate.client6.v1.internal.Proxy; import io.weaviate.client6.v1.internal.Timeout; import io.weaviate.client6.v1.internal.TokenProvider; @@ -13,27 +15,75 @@ public class GrpcChannelOptions extends TransportOptions { private final OptionalInt maxMessageSize; + private final Function channelBuilder; - public GrpcChannelOptions(String scheme, String host, int port, Map headers, - TokenProvider tokenProvider, TrustManagerFactory tmf, Timeout timeout, Proxy proxy) { - this(scheme, host, port, buildMetadata(headers), tokenProvider, tmf, OptionalInt.empty(), timeout, proxy); + public GrpcChannelOptions( + String scheme, + String host, + int port, + Map headers, + TokenProvider tokenProvider, + TrustManagerFactory tmf, + Timeout timeout, + Proxy proxy, + Function channelBuilder) { + this( + scheme, + host, + port, + buildMetadata(headers), + tokenProvider, + tmf, + OptionalInt.empty(), + timeout, + proxy, + channelBuilder); } - private GrpcChannelOptions(String scheme, String host, int port, Metadata headers, - TokenProvider tokenProvider, TrustManagerFactory tmf, OptionalInt maxMessageSize, Timeout timeout, Proxy proxy) { - super(scheme, host, port, headers, tokenProvider, tmf, timeout, proxy); + private GrpcChannelOptions(String scheme, + String host, + int port, + Metadata headers, + TokenProvider tokenProvider, + TrustManagerFactory tmf, + OptionalInt maxMessageSize, + Timeout timeout, + Proxy proxy, + Function channelBuilder) { + super( + scheme, + host, + port, + headers, + tokenProvider, + tmf, + timeout, + proxy); this.maxMessageSize = maxMessageSize; + this.channelBuilder = channelBuilder; } public GrpcChannelOptions withMaxMessageSize(int maxMessageSize) { - return new GrpcChannelOptions(scheme, host, port, headers, tokenProvider, trustManagerFactory, - OptionalInt.of(maxMessageSize), timeout, proxy); + return new GrpcChannelOptions(scheme, + host, + port, + headers, + tokenProvider, + trustManagerFactory, + OptionalInt.of(maxMessageSize), + timeout, + proxy, + channelBuilder); } public OptionalInt maxMessageSize() { return maxMessageSize; } + public Function channelBuilder() { + return channelBuilder; + } + private static final Metadata buildMetadata(Map headers) { var metadata = new Metadata(); for (var header : headers.entrySet()) { From a5b05203f5ae1431c55c0c79f48d81cac51fb825 Mon Sep 17 00:00:00 2001 From: dyma solovei Date: Mon, 26 Oct 2026 11:11:07 +0100 Subject: [PATCH 2/2] chore: fix grammar --- src/main/java/io/weaviate/client6/v1/api/Config.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/main/java/io/weaviate/client6/v1/api/Config.java b/src/main/java/io/weaviate/client6/v1/api/Config.java index dff690add..5c9872f7a 100644 --- a/src/main/java/io/weaviate/client6/v1/api/Config.java +++ b/src/main/java/io/weaviate/client6/v1/api/Config.java @@ -197,7 +197,8 @@ public SelfT proxy(Proxy proxy) { } /** - * Override or extend any of configurations previously set for the gRPC channel. + * Override or extend any of the configurations previously set for the gRPC + * channel. * *
* This is best used to access the parameters specific to gRPC, which the client