JCBC-2198, KCBC-192, SCBC-490: Operational SDK prevent connection to Analytics 2.0 Cluster

Use the prodName field introduced into the global and
bucket topologies, to prevent the operational SDK
performing operations against an Enterprise Analytics
cluster.

Change-Id: I945a7934b83e346dc6eb63ab4e756f1964fce07a
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/229751
Reviewed-by: Graham Pople <[email protected]>
Tested-by: Graham Pople <[email protected]>

Author: programmatix

core-io/src/main/java/com/couchbase/client/core/topology/ClusterIdentifier.java

@@ -15,6 +15,7 @@
  */
 package com.couchbase.client.core.topology;
 
+import com.couchbase.client.core.annotation.SinceCouchbase;
 import com.couchbase.client.core.annotation.Stability;
 import com.couchbase.client.core.deps.com.fasterxml.jackson.databind.JsonNode;
 import com.couchbase.client.core.deps.com.fasterxml.jackson.databind.node.ObjectNode;
@@ -26,19 +27,23 @@
 public class ClusterIdentifier {
   private final String clusterUuid;
   private final String clusterName;
+  @SinceCouchbase("8.0")
+  private final @Nullable String prodName;
 
-  ClusterIdentifier(String clusterUuid, String clusterName) {
+  ClusterIdentifier(String clusterUuid, String clusterName, @Nullable String prodName) {
     this.clusterUuid = clusterUuid;
     this.clusterName = clusterName;
+    this.prodName = prodName;
   }
 
   public static @Nullable ClusterIdentifier parse(ObjectNode config) {
     JsonNode clusterUuid = config.path("clusterUUID");
     JsonNode clusterName = config.path("clusterName");
+    JsonNode prodName = config.path("prodName");
     if (clusterUuid.isMissingNode() || clusterName.isMissingNode()) {
       return null;
     }
-    return new ClusterIdentifier(clusterUuid.asText(), clusterName.asText());
+    return new ClusterIdentifier(clusterUuid.asText(), clusterName.asText(), prodName.isMissingNode() ? null : prodName.asText());
   }
 
   public String clusterUuid() {
@@ -49,11 +54,16 @@ public String clusterName() {
     return clusterName;
   }
 
+  public @Nullable String prodName() {
+    return prodName;
+  }
+
   @Override
   public String toString() {
     return "ClusterIdent{" +
       "clusterUuid='" + clusterUuid + '\'' +
       ", clusterName='" + redactMeta(clusterName) + '\'' +
+      ", prodName='" + prodName + '\'' +
       '}';
   }
 }

core-io/src/main/java/com/couchbase/client/core/topology/ClusterProdName.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2025 Couchbase, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.couchbase.client.core.topology;
+
+import com.couchbase.client.core.annotation.Stability;
+
+// Known values for the prodName cluster field added in MB-67103
+@Stability.Internal
+public class ClusterProdName {
+    private ClusterProdName() {}
+
+    public static final String COUCHBASE_SERVER = "Couchbase Server";
+    public static final String ENTERPRISE_ANALYTICS = "Enterprise Analytics";
+}

core-io/src/main/java/com/couchbase/client/core/topology/ClusterTopologyBuilder.java

@@ -44,6 +44,7 @@ public final class ClusterTopologyBuilder {
   private NetworkResolution networkResolution = NetworkResolution.DEFAULT;
   private String clusterName = "fake-cluster";
   private String clusterUuid = "fake-cluster-uuid";
+  private String prodName = ClusterProdName.COUCHBASE_SERVER;
   private Set<ClusterCapability> capabilities = EnumSet.allOf(ClusterCapability.class);
   private final List<HostAndServicePorts> nodes = new ArrayList<>();
 
@@ -54,7 +55,7 @@ public ClusterTopology build() {
   private ClusterTopology buildWithOrWithoutBucket(@Nullable BucketTopology bucket) {
     return ClusterTopology.of(
       revision,
-      new ClusterIdentifier(clusterUuid, clusterName),
+      new ClusterIdentifier(clusterUuid, clusterName, prodName),
       nodes,
       capabilities,
       networkResolution,

java-client/src/main/java/com/couchbase/client/java/analytics/AnalyticsAccessor.java

@@ -18,8 +18,10 @@
 
 import com.couchbase.client.core.Core;
 import com.couchbase.client.core.Reactor;
+import com.couchbase.client.core.error.CouchbaseException;
 import com.couchbase.client.core.msg.analytics.AnalyticsRequest;
 import com.couchbase.client.core.msg.analytics.AnalyticsResponse;
+import com.couchbase.client.core.topology.ClusterProdName;
 import com.couchbase.client.java.codec.JsonSerializer;
 import reactor.core.publisher.Mono;
 
@@ -52,11 +54,28 @@ public static Mono<ReactiveAnalyticsResult> analyticsQueryReactive(final Core co
   }
 
   private static Mono<AnalyticsResponse> analyticsQueryInternal(final Core core, final AnalyticsRequest request) {
-    core.send(request);
-    return Reactor
-      .wrap(request, request.response(), true)
-      .doOnNext(ignored -> request.context().logicallyComplete())
-      .doOnError(err -> request.context().logicallyComplete(err));
+    return core.waitForClusterTopology(request.timeout())
+                    .flatMap(clusterTopology -> {
+
+                      if (clusterTopology.id() != null
+                              && clusterTopology.id().prodName() != null
+                              && !clusterTopology.id().prodName().startsWith(ClusterProdName.COUCHBASE_SERVER)) {
+                        StringBuilder sb = new StringBuilder();
+                        sb.append("This '");
+                        sb.append(clusterTopology.id().prodName());
+                        sb.append("' cluster cannot be used with this SDK, which is intended for use with operational clusters");
+                        if (clusterTopology.id().prodName().startsWith(ClusterProdName.ENTERPRISE_ANALYTICS)) {
+                          sb.append(". For this cluster, an Enterprise Analytics SDK should be used.");
+                        }
+                        return Mono.error(new CouchbaseException(sb.toString()));
+                      }
+
+                      return Mono.defer(() -> {
+                        core.send(request);
+                        return Reactor.wrap(request, request.response(), true);
+                      }).doOnNext(ignored -> request.context().logicallyComplete())
+                        .doOnError(err -> request.context().logicallyComplete(err));
+                    });
   }
 
 }