fix(batch): reject servers older than 1.36 in batch.start()

src/main/java/io/weaviate/client6/v1/api/WeaviateClient.java

@@ -120,6 +120,7 @@ public WeaviateClient(Config config) {
       throw new WeaviateUnsupportedVersionException(meta.version());
     }
 
+    grpcOpt = grpcOpt.withServerVersion(meta.version());
     if (meta.grpcMaxMessageSize() != null) {
       grpcOpt = grpcOpt.withMaxMessageSize(meta.grpcMaxMessageSize());
     }

src/main/java/io/weaviate/client6/v1/api/WeaviateClientAsync.java

@@ -124,6 +124,7 @@ public WeaviateClientAsync(Config config) {
       throw new WeaviateUnsupportedVersionException(meta.version());
     }
 
+    grpcOpt = grpcOpt.withServerVersion(meta.version());
     if (meta.grpcMaxMessageSize() != null) {
       grpcOpt = grpcOpt.withMaxMessageSize(meta.grpcMaxMessageSize());
     }

src/main/java/io/weaviate/client6/v1/api/collections/batch/WeaviateBatchClient.java

@@ -8,6 +8,7 @@
 import io.weaviate.client6.v1.api.collections.CollectionHandleDefaults;
 import io.weaviate.client6.v1.internal.ObjectBuilder;
 import io.weaviate.client6.v1.internal.TransportOptions;
+import io.weaviate.client6.v1.internal.VersionSupport;
 import io.weaviate.client6.v1.internal.grpc.GrpcTransport;
 import io.weaviate.client6.v1.internal.orm.CollectionDescriptor;
 
@@ -34,14 +35,11 @@ public WeaviateBatchClient(WeaviateBatchClient<PropertiesT> c, CollectionHandleD
 
   public BatchContext<PropertiesT> start(
       Function<BatchContext.Builder<PropertiesT>, ObjectBuilder<BatchContext<PropertiesT>>> fn) {
-    OptionalInt maxSizeBytes = grpcTransport.maxMessageSizeBytes();
-    if (maxSizeBytes.isEmpty()) {
-      throw new IllegalStateException("Server must have grpcMaxMessageSize configured to use server-side batching");
-    }
+    int maxSizeBytes = requireServerSideBatching();
 
     StreamFactory<Message, Event> streamFactory = new TranslatingStreamFactory(grpcTransport::createStream);
     BatchContext.Builder<PropertiesT> builder = new BatchContext.Builder<>(
-        streamFactory, maxSizeBytes.getAsInt(), collectionDescriptor, defaults);
+        streamFactory, maxSizeBytes, collectionDescriptor, defaults);
     BatchContext<PropertiesT> context = fn.apply(builder).build();
 
     if (isWeaviateCloudOnGoogleCloud(grpcTransport.host())) {
@@ -53,15 +51,12 @@ public BatchContext<PropertiesT> start(
   }
 
   public BatchContext<PropertiesT> start() {
-    OptionalInt maxSizeBytes = grpcTransport.maxMessageSizeBytes();
-    if (maxSizeBytes.isEmpty()) {
-      throw new IllegalStateException("Server must have grpcMaxMessageSize configured to use server-side batching");
-    }
+    int maxSizeBytes = requireServerSideBatching();
 
     StreamFactory<Message, Event> streamFactory = new TranslatingStreamFactory(grpcTransport::createStream);
     BatchContext<PropertiesT> context = new BatchContext.Builder<>(
         streamFactory,
-        maxSizeBytes.getAsInt(),
+        maxSizeBytes,
         collectionDescriptor,
         defaults).build();
 
@@ -73,6 +68,22 @@ public BatchContext<PropertiesT> start() {
     return context;
   }
 
+  private int requireServerSideBatching() {
+    grpcTransport.serverVersion().ifPresent(version -> {
+      if (!VersionSupport.isAtLeast(version, VersionSupport.SERVER_SIDE_BATCHING_VERSION)) {
+        throw new IllegalStateException(
+            "Server-side batching requires Weaviate %s or newer, server version is %s"
+                .formatted(VersionSupport.SERVER_SIDE_BATCHING_VERSION, version));
+      }
+    });
+
+    OptionalInt maxSizeBytes = grpcTransport.maxMessageSizeBytes();
+    if (maxSizeBytes.isEmpty()) {
+      throw new IllegalStateException("Server must have grpcMaxMessageSize configured to use server-side batching");
+    }
+    return maxSizeBytes.getAsInt();
+  }
+
   private static final int GCP_RECONNECT_INTERVAL_SECONDS = 160;
 
   private static boolean isWeaviateCloudOnGoogleCloud(String host) {

src/main/java/io/weaviate/client6/v1/internal/VersionSupport.java

@@ -4,6 +4,7 @@
 
 public final class VersionSupport {
   public static final SemanticVersion MINIMAL_SUPPORTED_VERSION = new SemanticVersion(1, 32);
+  public static final SemanticVersion SERVER_SIDE_BATCHING_VERSION = new SemanticVersion(1, 36);
 
   /**
    * Returns true if the {@code version} is the same as or older than the
@@ -14,6 +15,13 @@ public static boolean isSupported(String version) {
     return semver.compareTo(MINIMAL_SUPPORTED_VERSION) >= 0;
   }
 
+  public static boolean isAtLeast(String version, SemanticVersion minimal) {
+    var semver = SemanticVersion.of(version);
+    return semver.major() != minimal.major()
+        ? semver.major() > minimal.major()
+        : semver.minor() >= minimal.minor();
+  }
+
   public record SemanticVersion(int major, int minor, String patch) implements Comparable<SemanticVersion> {
 
     public SemanticVersion(int major, int minor) {

src/main/java/io/weaviate/client6/v1/internal/grpc/DefaultGrpcTransport.java

@@ -4,6 +4,7 @@
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
@@ -75,6 +76,11 @@ public OptionalInt maxMessageSizeBytes() {
     return transportOptions.maxMessageSize();
   }
 
+  @Override
+  public Optional<String> serverVersion() {
+    return transportOptions.serverVersion();
+  }
+
   @Override
   public <RequestT, RequestM, ReplyM, ResponseT> ResponseT performRequest(RequestT request,
       Rpc<RequestT, RequestM, ResponseT, ReplyM> rpc) {

src/main/java/io/weaviate/client6/v1/internal/grpc/GrpcChannelOptions.java

@@ -1,6 +1,7 @@
 package io.weaviate.client6.v1.internal.grpc;
 
 import java.util.Map;
+import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.function.Function;
 
@@ -15,6 +16,7 @@
 
 public class GrpcChannelOptions extends TransportOptions<Metadata> {
   private final OptionalInt maxMessageSize;
+  private final String serverVersion;
   private final Function<NettyChannelBuilder, NettyChannelBuilder> channelBuilder;
 
   public GrpcChannelOptions(
@@ -35,6 +37,7 @@ public GrpcChannelOptions(
         tokenProvider,
         tmf,
         OptionalInt.empty(),
+        null,
         timeout,
         proxy,
         channelBuilder);
@@ -47,6 +50,7 @@ private GrpcChannelOptions(String scheme,
       TokenProvider tokenProvider,
       TrustManagerFactory tmf,
       OptionalInt maxMessageSize,
+      String serverVersion,
       Timeout timeout,
       Proxy proxy,
       Function<NettyChannelBuilder, NettyChannelBuilder> channelBuilder) {
@@ -60,6 +64,7 @@ private GrpcChannelOptions(String scheme,
         timeout,
         proxy);
     this.maxMessageSize = maxMessageSize;
+    this.serverVersion = serverVersion;
     this.channelBuilder = channelBuilder;
   }
 
@@ -71,6 +76,7 @@ public GrpcChannelOptions withMaxMessageSize(int maxMessageSize) {
         tokenProvider,
         trustManagerFactory,
         OptionalInt.of(maxMessageSize),
+        serverVersion,
         timeout,
         proxy,
         channelBuilder);
@@ -80,6 +86,24 @@ public OptionalInt maxMessageSize() {
     return maxMessageSize;
   }
 
+  public GrpcChannelOptions withServerVersion(String serverVersion) {
+    return new GrpcChannelOptions(scheme,
+        host,
+        port,
+        headers,
+        tokenProvider,
+        trustManagerFactory,
+        maxMessageSize,
+        serverVersion,
+        timeout,
+        proxy,
+        channelBuilder);
+  }
+
+  public Optional<String> serverVersion() {
+    return Optional.ofNullable(serverVersion);
+  }
+
   public Function<NettyChannelBuilder, NettyChannelBuilder> channelBuilder() {
     return channelBuilder;
   }

src/main/java/io/weaviate/client6/v1/internal/grpc/GrpcTransport.java

@@ -1,5 +1,6 @@
 package io.weaviate.client6.v1.internal.grpc;
 
+import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.concurrent.CompletableFuture;
 
@@ -30,4 +31,8 @@ StreamObserver<WeaviateProtoBatch.BatchStreamRequest> createStream(
    * Maximum inbound/outbound message size supported by the underlying channel.
    */
   OptionalInt maxMessageSizeBytes();
+
+  default Optional<String> serverVersion() {
+    return Optional.empty();
+  }
 }

src/test/java/io/weaviate/client6/v1/api/collections/batch/WeaviateBatchClientTest.java

@@ -0,0 +1,25 @@
+package io.weaviate.client6.v1.api.collections.batch;
+
+import java.util.Map;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import io.weaviate.client6.v1.api.collections.CollectionHandleDefaults;
+import io.weaviate.client6.v1.internal.orm.CollectionDescriptor;
+import io.weaviate.testutil.transport.MockGrpcTransport;
+
+public class WeaviateBatchClientTest {
+  @Test
+  public void test_start_rejectsServerBefore136() {
+    var batch = new WeaviateBatchClient<Map<String, Object>>(
+        new MockGrpcTransport("1.35.6"),
+        CollectionDescriptor.ofMap("Things"),
+        CollectionHandleDefaults.of(CollectionHandleDefaults.none()));
+
+    Assertions.assertThatThrownBy(batch::start)
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("1.36")
+        .hasMessageContaining("1.35.6");
+  }
+}

src/test/java/io/weaviate/client6/v1/internal/VersionSupportTest.java

@@ -37,4 +37,22 @@ public void test_isSupported(String minimal, String actual, boolean isSupported)
           .isLessThan(v_minimal);
     }
   }
+
+  public static Object[][] isAtLeastCases() {
+    return new Object[][] {
+        { "1.35.6", false },
+        { "1.36.0", true },
+        { "1.36.0-rc.1", true },
+        { "1.39.7", true },
+        { "2.0.1", true },
+    };
+  }
+
+  @Test
+  @DataMethod(source = VersionSupportTest.class, method = "isAtLeastCases")
+  public void test_isAtLeast(String version, boolean want) {
+    Assertions.assertThat(VersionSupport.isAtLeast(version, VersionSupport.SERVER_SIDE_BATCHING_VERSION))
+        .describedAs("%s >= %s", version, VersionSupport.SERVER_SIDE_BATCHING_VERSION)
+        .isEqualTo(want);
+  }
 }

src/test/java/io/weaviate/testutil/transport/MockGrpcTransport.java

@@ -3,6 +3,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.concurrent.CompletableFuture;
 
@@ -19,6 +20,15 @@
 public class MockGrpcTransport implements GrpcTransport {
   private final String host = "example.com";
   private final static int maxMessageSizeBytes = 10240; // 10KB
+  private final String serverVersion;
+
+  public MockGrpcTransport() {
+    this(null);
+  }
+
+  public MockGrpcTransport(String serverVersion) {
+    this.serverVersion = serverVersion;
+  }
 
   @FunctionalInterface
   public interface AssertFunction {
@@ -74,6 +84,11 @@ public OptionalInt maxMessageSizeBytes() {
     return OptionalInt.of(maxMessageSizeBytes);
   }
 
+  @Override
+  public Optional<String> serverVersion() {
+    return Optional.ofNullable(serverVersion);
+  }
+
   @Override
   public String host() {
     return host;