This is an automated email from the ASF dual-hosted git repository.
tbonelee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new 933da6862c [ZEPPELIN-6531] Fail fast when ZeppelinWebSocketClient
cannot connect
933da6862c is described below
commit 933da6862c18240002c8e930b9148a8ab96402bb
Author: HwangRock <[email protected]>
AuthorDate: Tue Jul 14 22:43:41 2026 +0900
[ZEPPELIN-6531] Fail fast when ZeppelinWebSocketClient cannot connect
### What is this PR for?
`ZeppelinWebSocketClient.connect()` blocks forever when the websocket
handshake fails. It waits on `connectLatch.await()` with no timeout, and the
latch is only counted down in the success callback (`onConnect`). Pre-handshake
failures — connection refused, TLS error, rejected upgrade — are not delivered
through the annotated callbacks at all; Jetty completes the `CompletableFuture`
returned by `WebSocketClient.connect()` exceptionally instead. The original
code discards that future, [...]
This affects any `zeppelin-client` (`ZSession`) user that passes a
`MessageHandler` to `start()`/`reconnect()`: if the target Zeppelin server is
down or unreachable, the calling thread blocks indefinitely.
Fix: wait on the future with a bounded timeout and surface handshake
failures as `IOException`. On failure, stop the Jetty client so its threads are
not leaked. `connectLatch` is unused after this change, so it is removed
together with its unused `getConnectLatch()` getter.
Note on approach: adding `connectLatch.countDown()` to `onError` does not
fix this. Against jetty-websocket 11, pre-handshake failures reach only the
future, never `<at>OnWebSocketError`, so the latch would still never be
released. The future is the only path that carries the failure.
### What type of PR is it?
Bug Fix
### Todos
* [ ] - Follow-up: make the connect timeout configurable via `ClientConfig`
(currently a 30s constant)
### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-6531
### How should this be tested?
Added `ZeppelinWebSocketClientTest` (the first unit test in
`zeppelin-client`): connecting to a closed port must fail within a bounded time
instead of hanging.
Also verified end-to-end against a docker-exposed port. Ran `nginx` on a
port, which accepts the TCP connection but rejects the websocket upgrade (HTTP
404 on `/ws`), and pointed the client at both that port and a closed port.
Before (original code, same docker port):
```
[INFO] Running org.apache.zeppelin.client.websocket.ManualE2EVerifyTest
```
Hangs here — no result line. The forked surefire JVM stayed alive past 45s
and had to be killed.
After (this PR):
```
[E2E] closed-port(1) failed-fast in 295ms -> IOException: Failed to
establish websocket connection to ws://127.0.0.1:1/ws
[E2E] open-port-non-ws(18080) failed-fast in 45ms -> IOException: Failed to
establish websocket connection to ws://127.0.0.1:18080/ws
```
The docker probe was a throwaway check and is not part of the PR; the
committed test uses a closed port and needs no docker.
### Screenshots (if appropriate)
### Questions:
* Does the license files need to update? No.
* Is there breaking changes for older versions? `getConnectLatch()` is
removed. It is public but has no callers in the repo, and it only exposed
`connect()`'s internal latch, which no longer exists.
* Does this needs documentation? No.
Closes #5300 from HwangRock/ZEPPELIN-6531.
Signed-off-by: ChanHo Lee <[email protected]>
---
.../client/websocket/ZeppelinWebSocketClient.java | 35 ++++++++++++++------
.../websocket/ZeppelinWebSocketClientTest.java | 37 ++++++++++++++++++++++
2 files changed, 62 insertions(+), 10 deletions(-)
diff --git
a/zeppelin-client/src/main/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClient.java
b/zeppelin-client/src/main/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClient.java
index a3e700398a..c09fb845ad 100644
---
a/zeppelin-client/src/main/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClient.java
+++
b/zeppelin-client/src/main/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClient.java
@@ -33,8 +33,11 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.URI;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
/**
@@ -45,8 +48,8 @@ import java.util.concurrent.TimeUnit;
public class ZeppelinWebSocketClient {
private static final Logger LOGGER =
LoggerFactory.getLogger(ZeppelinWebSocketClient.class);
private static final Gson GSON = new Gson();
+ private static final long DEFAULT_CONNECT_TIMEOUT_MS = 30_000;
- private CountDownLatch connectLatch = new CountDownLatch(1);
private CountDownLatch closeLatch = new CountDownLatch(1);
private Session session;
@@ -63,8 +66,18 @@ public class ZeppelinWebSocketClient {
URI echoUri = new URI(url);
ClientUpgradeRequest request = new ClientUpgradeRequest();
request.setHeader("Origin", "*");
- wsClient.connect(this, echoUri, request);
- connectLatch.await();
+ CompletableFuture<Session> future = wsClient.connect(this, echoUri,
request);
+ try {
+ future.get(DEFAULT_CONNECT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ stopQuietly();
+ throw new IOException("Timeout(" + DEFAULT_CONNECT_TIMEOUT_MS
+ + "ms) establishing websocket connection to " + url, e);
+ } catch (ExecutionException e) {
+ stopQuietly();
+ throw new IOException("Failed to establish websocket connection to " +
url,
+ e.getCause());
+ }
LOGGER.info("WebSocket connect established");
}
@@ -93,7 +106,6 @@ public class ZeppelinWebSocketClient {
public void onConnect(Session session) {
LOGGER.info("Got connect: {}", session.getRemote());
this.session = session;
- connectLatch.countDown();
}
@OnWebSocketMessage
@@ -103,22 +115,25 @@ public class ZeppelinWebSocketClient {
@OnWebSocketError
public void onError(Throwable cause) {
- LOGGER.info("WebSocket Error: " + cause.getMessage());
- cause.printStackTrace(System.out);
+ LOGGER.error("WebSocket error", cause);
}
public void send(Message message) throws IOException {
session.getRemote().sendString(GSON.toJson(message));
}
- public CountDownLatch getConnectLatch() {
- return connectLatch;
- }
-
public void stop() throws Exception {
if (this.wsClient != null) {
this.wsClient.stop();
}
}
+ private void stopQuietly() {
+ try {
+ stop();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to stop websocket client after connection failure",
e);
+ }
+ }
+
}
diff --git
a/zeppelin-client/src/test/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClientTest.java
b/zeppelin-client/src/test/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClientTest.java
new file mode 100644
index 0000000000..7f6cabd1e2
--- /dev/null
+++
b/zeppelin-client/src/test/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClientTest.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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 org.apache.zeppelin.client.websocket;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+
+class ZeppelinWebSocketClientTest {
+
+ @Test
+ void connectFailsFastWhenPortClosed() {
+ ZeppelinWebSocketClient client = new ZeppelinWebSocketClient(msg -> { });
+
+ assertTimeoutPreemptively(Duration.ofSeconds(20), () ->
+ assertThrows(Exception.class, () ->
client.connect("ws://127.0.0.1:1/ws")));
+ }
+
+}
