JVMCBC-1697 Global KV connections ignore negotiated features

Motivation
----------
KeyValueMessageHandler.channelActive() was accessing the "negotiated
features" attribute before FeatureNegotiatingHandler set it.

Modifications
-------------
Insert a barrier that absorbs the channelActive event from pipelined
handshake handlers, and defers it until the handshake is complete.

Change-Id: I2ad233649e92b7d47eacc416a563e54e7f8a1583
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/235399
Reviewed-by: Michael Reiche <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: dnault

core-io/src/main/java/com/couchbase/client/core/endpoint/KeyValueEndpoint.java

@@ -22,6 +22,7 @@
 import com.couchbase.client.core.io.netty.TrafficCaptureHandler;
 import com.couchbase.client.core.io.netty.kv.ErrorMapLoadingHandler;
 import com.couchbase.client.core.io.netty.kv.FeatureNegotiatingHandler;
+import com.couchbase.client.core.io.netty.kv.HandshakeBarrier;
 import com.couchbase.client.core.io.netty.kv.KeyValueMessageHandler;
 import com.couchbase.client.core.io.netty.kv.MemcacheProtocolDecodeHandler;
 import com.couchbase.client.core.io.netty.kv.MemcacheProtocolVerificationHandler;
@@ -105,6 +106,7 @@ public void init(BaseEndpoint endpoint, ChannelPipeline pipeline) {
       authenticator.authKeyValueConnection(ctx, pipeline);
 
       bucketname.ifPresent(s -> pipeline.addLast(new SelectBucketHandler(ctx, s)));
+      pipeline.addLast(new HandshakeBarrier());
       pipeline.addLast(new KeyValueMessageHandler(endpoint, ctx, bucketname));
     }
 

core-io/src/main/java/com/couchbase/client/core/io/netty/kv/HandshakeBarrier.java

@@ -0,0 +1,77 @@
+/*
+ * 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.io.netty.kv;
+
+import com.couchbase.client.core.deps.io.netty.channel.ChannelDuplexHandler;
+import com.couchbase.client.core.deps.io.netty.channel.ChannelHandlerContext;
+import com.couchbase.client.core.deps.io.netty.channel.ChannelPromise;
+
+import java.net.SocketAddress;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Waits for the pipelined handshake sequence to complete before propagating
+ * the `channelActive` signal.
+ * <p>
+ * Necessary because {@link KeyValueMessageHandler#channelActive}
+ * depends on channel attributes set by pipelined handshake handlers.
+ * <p>
+ * @implNote Some handshake handlers are implicit barriers
+ * (specifically {@link SelectBucketHandler} and {@link SaslAuthenticationHandler}
+ * with a multistep mechanism), but they are not always present.
+ */
+public class HandshakeBarrier extends ChannelDuplexHandler {
+
+  private ChannelPromise interceptedConnectPromise;
+
+  @Override
+  public void connect(
+    ChannelHandlerContext ctx,
+    SocketAddress remoteAddress,
+    SocketAddress localAddress,
+    ChannelPromise promise
+  ) {
+    interceptedConnectPromise = requireNonNull(promise);
+
+    ChannelPromise downstream = ctx.newPromise();
+    downstream.addListener(f -> {
+      if (f.isSuccess()) {
+        interceptedConnectPromise.trySuccess();
+        ctx.pipeline().remove(this);
+        ctx.fireChannelActive();
+
+      } else {
+        interceptedConnectPromise.tryFailure(f.cause());
+      }
+    });
+
+    ctx.connect(remoteAddress, localAddress, downstream);
+  }
+
+  @Override
+  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+    interceptedConnectPromise.tryFailure(cause);
+    ctx.fireExceptionCaught(cause);
+  }
+
+  @Override
+  public void channelActive(ChannelHandlerContext ctx) {
+    // Prevent this signal from reaching non-pipelined handlers,
+    // because the pipelined handshake handlers might still be waiting for responses.
+  }
+}