Fixes for EE11 Jakarta WebSocket implementation to fix TCK failures.

jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/HttpFieldsMap.java

@@ -98,6 +98,14 @@ public List<String> remove(Object key)
             return null;
         }
 
+        @Override
+        public boolean containsKey(Object key)
+        {
+            if (key instanceof String s)
+                return httpFields.contains(s);
+            return false;
+        }
+
         @Override
         public Set<Entry<String, List<String>>> entrySet()
         {
@@ -178,6 +186,14 @@ public List<String> remove(Object key)
             throw new UnsupportedOperationException();
         }
 
+        @Override
+        public boolean containsKey(Object key)
+        {
+            if (key instanceof String s)
+                return httpFields.contains(s);
+            return false;
+        }
+
         @Override
         public Set<Entry<String, List<String>>> entrySet()
         {

jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/messages/AbstractMessageSink.java

@@ -42,7 +42,7 @@
 public abstract class AbstractMessageSink implements MessageSink
 {
     private final CoreSession session;
-    private final MethodHolder methodHandle;
+    private final MethodHolder methodHolder;
     private final boolean autoDemand;
 
     /**
@@ -56,7 +56,7 @@ public abstract class AbstractMessageSink implements MessageSink
     public AbstractMessageSink(CoreSession session, MethodHolder methodHolder, boolean autoDemand)
     {
         this.session = Objects.requireNonNull(session, "CoreSession");
-        this.methodHandle = Objects.requireNonNull(methodHolder, "MethodHolder");
+        this.methodHolder = Objects.requireNonNull(methodHolder, "MethodHolder");
         this.autoDemand = autoDemand;
     }
 
@@ -75,7 +75,7 @@ public CoreSession getCoreSession()
      */
     public MethodHolder getMethodHolder()
     {
-        return methodHandle;
+        return methodHolder;
     }
 
     /**

jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/messages/DispatchedMessageSink.java

@@ -19,6 +19,8 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
 
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.IO;
@@ -48,8 +50,10 @@
 public abstract class DispatchedMessageSink extends AbstractMessageSink
 {
     private final Executor executor;
+    private final AtomicBoolean wasCallbackFailed = new AtomicBoolean(false);
     private volatile CompletableFuture<Void> dispatchComplete;
     private MessageSink typeSink;
+    private Consumer<Throwable> onError;
 
     public DispatchedMessageSink(CoreSession session, MethodHolder methodHolder, boolean autoDemand)
     {
@@ -59,6 +63,15 @@ public DispatchedMessageSink(CoreSession session, MethodHolder methodHolder, boo
         executor = session.getWebSocketComponents().getExecutor();
     }
 
+    public DispatchedMessageSink(CoreSession session, MethodHolder methodHolder, boolean autoDemand, Consumer<Throwable> onError)
+    {
+        super(session, methodHolder, autoDemand);
+        if (!autoDemand)
+            throw new IllegalArgumentException("%s must be auto-demanding".formatted(getClass().getSimpleName()));
+        this.executor = session.getWebSocketComponents().getExecutor();
+        this.onError = onError;
+    }
+
     public abstract MessageSink newMessageSink();
 
     public void accept(Frame frame, final Callback callback)
@@ -105,17 +118,31 @@ public void accept(Frame frame, final Callback callback)
                 {
                     autoDemand();
                 }
-                else
+                // We only need to handle the error here if none of the callbacks were ever failed.
+                else if (!wasCallbackFailed.get())
                 {
                     if (failure instanceof CompletionException completionException)
                         failure = completionException.getCause();
 
-                    CloseStatus closeStatus = new CloseStatus(CloseStatus.SERVER_ERROR, failure);
-                    getCoreSession().close(closeStatus, Callback.NOOP);
+                    if (onError == null)
+                    {
+                        CloseStatus closeStatus = new CloseStatus(CloseStatus.SERVER_ERROR, failure);
+                        getCoreSession().close(closeStatus, Callback.NOOP);
+                    }
+                    else
+                    {
+                        onError.accept(failure);
+                    }
                 }
             });
         }
 
+        frameCallback = Callback.from(frameCallback::succeeded, throwable ->
+        {
+            if (throwable != null)
+                wasCallbackFailed.set(true);
+            callback.failed(throwable);
+        });
         typeSink.accept(frame, frameCallback);
     }
 

jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/messages/InputStreamMessageSink.java

@@ -13,6 +13,8 @@
 
 package org.eclipse.jetty.websocket.core.messages;
 
+import java.util.function.Consumer;
+
 import org.eclipse.jetty.websocket.core.CoreSession;
 import org.eclipse.jetty.websocket.core.util.MethodHolder;
 
@@ -23,6 +25,11 @@ public InputStreamMessageSink(CoreSession session, MethodHolder methodHolder, bo
         super(session, methodHolder, autoDemand);
     }
 
+    public InputStreamMessageSink(CoreSession session, MethodHolder methodHolder, boolean autoDemand, Consumer<Throwable> onError)
+    {
+        super(session, methodHolder, autoDemand, onError);
+    }
+
     @Override
     public MessageSink newMessageSink()
     {

jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/messages/MessageOutputStream.java

@@ -171,6 +171,9 @@ public void close() throws IOException
     {
         try
         {
+            if (closed)
+                return;
+
             flush(true);
             buffer.release();
             if (LOG.isDebugEnabled())

jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/messages/ReaderMessageSink.java

@@ -13,6 +13,8 @@
 
 package org.eclipse.jetty.websocket.core.messages;
 
+import java.util.function.Consumer;
+
 import org.eclipse.jetty.websocket.core.CoreSession;
 import org.eclipse.jetty.websocket.core.util.MethodHolder;
 
@@ -23,6 +25,11 @@ public ReaderMessageSink(CoreSession session, MethodHolder methodHolder, boolean
         super(session, methodHolder, autoDemand);
     }
 
+    public ReaderMessageSink(CoreSession session, MethodHolder methodHolder, boolean autoDemand, Consumer<Throwable> onError)
+    {
+        super(session, methodHolder, autoDemand, onError);
+    }
+
     @Override
     public MessageReader newMessageSink()
     {