You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@zeppelin.apache.org by "JangAyeon (via GitHub)" <gi...@apache.org> on 2026/09/24 09:44:44 UTC
[PR] [ZEPPELIN-6694] Reap dead websocket connections via pong tracking [zeppelin]
JangAyeon opened a new pull request, #5500:
URL: https://github.com/apache/zeppelin/pull/5500
### What is this PR for?
ZEPPELIN-6092 added server-initiated websocket ping frames to keep connections alive, but the returning pong frames are not tracked. Since every server-side write resets Jetty's idle timer, the idle timeout no longer detects dead clients once the heartbeat is on, and a dead session stays open until the OS gives up on TCP retransmission.
This PR restores liveness detection explicitly:
- Register a per-session `PongMessage` handler in `NotebookServer#onOpen`.
- Track unanswered pings and the last pong timestamp in `NotebookSocket`. A pong resets the counter; a failed ping write still counts as unanswered.
- In `sendHeartbeat`, close a session that has missed N consecutive pongs instead of pinging it again. The connection is removed from `sessionIdNotebookSocketMap` and `ConnectionManager` before closing, because a dead peer may never complete the close handshake (so `onClose` may arrive late or not at all).
- Add `zeppelin.websocket.heartbeat.max.missed.pongs` (default `3`, `<= 0` disables reaping). With the default 60s interval, a dead client is reaped within ~3-4 minutes, below the 300s default idle timeout.
- Sessions are closed with `GOING_AWAY` (1001), matching the code Jetty uses for idle timeouts.
### What type of PR is it?
Improvement
### Todos
* [x] - Track pong frames per websocket session and reset the counter on each pong
* [x] - Close sessions that miss N consecutive pongs in the heartbeat loop
* [x] - Add `zeppelin.websocket.heartbeat.max.missed.pongs` config (default 3, <= 0 disables)
* [x] - Update docs and `zeppelin-site.xml.template`
* [x] - Add unit tests
### What is the Jira issue?
[ZEPPELIN-6694](https://issues.apache.org/jira/browse/ZEPPELIN-6694)
### How should this be tested?
* Unit tests added:
* `NotebookSocketTest`: unanswered ping counting, reset on pong, failed ping still counted, close swallows exceptions
* `NotebookServerHeartbeatTest`: a session with too many missed pongs is closed and removed from `ConnectionManager`, a healthy session keeps receiving pings, nothing is closed when reaping is disabled
* `ZeppelinConfigurationTest`: default and override values for the new config
* Run: `./mvnw -pl zeppelin-server test -Dtest='NotebookSocketTest,NotebookServerHeartbeatTest,ZeppelinConfigurationTest'`
### Questions:
* Does the license files need to update? No
* Is there breaking changes for older versions? No. Reaping is on by default, but only affects sessions that stop answering pings, and can be disabled by setting the new property to 0.
* Does this needs documentation? Yes, included in `docs/setup/operation/configuration.md`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6694] Reap dead websocket connections via pong tracking [zeppelin]
Posted by "JangAyeon (via GitHub)" <gi...@apache.org>.
JangAyeon commented on PR #5500:
URL: https://github.com/apache/zeppelin/pull/5500#issuecomment-5826266539
Thanks for the thorough review, @tbonelee! I've addressed all comments and amended them into the original commit.
**Reaping during long-running ops**
Added a `handlingMessage` flag to `NotebookSocket`. `onMessage` sets it in a try/finally, and `sendHeartbeat` skips reaping while it is set, so queued pongs are no longer mistaken for missed ones.
**Test for the pong handler path**
Added `onOpenRegistersPongHandlerThatResetsPingCount`. It goes through `onOpen(Session, EndpointConfig)`, captures the registered handler, and checks that it resets the ping count. It now fails if the registration line is removed.
**Cleanups**
- Renamed `getUnansweredPings` to `getPingsSinceLastPong` and dropped its Javadoc
- Removed the comment above the pong handler
- Trimmed the config comment to what the setting means
- Dropped the session id null check and stubbed `getSessionId()` in the test
I also added tests for skipping busy sessions and for the `handlingMessage` flag, and updated the PR description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6694] Reap dead websocket connections via pong tracking [zeppelin]
Posted by "tbonelee (via GitHub)" <gi...@apache.org>.
tbonelee merged PR #5500:
URL: https://github.com/apache/zeppelin/pull/5500
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6694] Reap dead websocket connections via pong tracking [zeppelin]
Posted by "tbonelee (via GitHub)" <gi...@apache.org>.
tbonelee commented on code in PR #5500:
URL: https://github.com/apache/zeppelin/pull/5500#discussion_r4092546828
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -251,6 +252,8 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
if (checkOrigin(origin)) {
NotebookSocket notebookSocket = sessionIdNotebookSocketMap
.computeIfAbsent(session.getId(), unused -> new NotebookSocket(session, headers));
+ // Per-session pong handler so the heartbeat can tell live peers from dead ones.
Review Comment:
We could remove unnecessary comment.
```suggestion
```
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookSocket.java:
##########
@@ -67,15 +79,48 @@ public void send(String serializeMessage) throws IOException {
* session resets Jetty's idle timeout as well as any intermediate proxy's idle timer, so no
* application-level handling is required on the client. Exceptions are swallowed and logged
* so a single dead session cannot break the caller's heartbeat loop over all sessions.
+ * Every call counts as one outstanding ping until {@link #onPong()} is called, including
+ * calls whose write failed, since a failed write is itself a sign the peer is gone.
*/
public void sendPing() {
+ unansweredPings.incrementAndGet();
try {
session.getBasicRemote().sendPing(PING_PAYLOAD);
} catch (IOException | IllegalArgumentException | IllegalStateException e) {
LOGGER.warn("Failed to send heartbeat ping to session {}: {}", session.getId(), e.toString());
}
}
+ /**
+ * Records a pong frame from the peer. Any pong proves the connection is alive, so the
+ * outstanding-ping counter is reset rather than decremented.
+ */
+ public void onPong() {
+ lastPongTimestamp = System.currentTimeMillis();
+ unansweredPings.set(0);
+ }
+
+ /** Number of pings sent since the last pong was received. */
+ public int getUnansweredPings() {
Review Comment:
We could let the method name carry this and drop the comment. (`sendHeartbeat()` and the tests would need the same rename.)
```suggestion
public int getPingsSinceLastPong() {
```
##########
zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java:
##########
@@ -1105,6 +1109,11 @@ public enum ConfVars {
// per-connection traffic low. 60s gives 5 pings within the 300s default idle window.
// <= 0 disables server-initiated heartbeats.
ZEPPELIN_WEBSOCKET_HEARTBEAT_INTERVAL("zeppelin.websocket.heartbeat.interval", 60000L),
+ // Number of consecutive heartbeat pings left unanswered (no pong) before the server closes
+ // the session as dead. Needed because the heartbeat itself keeps resetting the Jetty idle
+ // timer, so the idle timeout can no longer detect dead clients (ZEPPELIN-6694). With the
+ // default 60s interval, a dead client is reaped after ~3-4 minutes. <= 0 disables reaping.
Review Comment:
The rationale (the heartbeat keeps resetting the Jetty idle timer) is also in the `sendHeartbeat()` Javadoc. We could keep it in one place and leave only what the setting means here.
```suggestion
// Consecutive unanswered pings before a session is closed as dead. <= 0 disables reaping.
```
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -319,19 +322,45 @@ synchronized void stopHeartbeatScheduler() {
* point of this heartbeat: it keeps connections alive even when the client-side application
* keep-alive timer is throttled or stopped (e.g. a backgrounded browser tab). A single
* session failing to receive a ping must not stop the remaining sessions from being pinged.
- * Pong responses are not tracked; once the heartbeat is enabled, Jetty's idle timeout no
- * longer determines connection liveness (see ZEPPELIN-6694).
+ *
+ * <p>Because these writes keep resetting Jetty's idle timer, the idle timeout can no longer
+ * detect dead clients. Liveness is therefore tracked explicitly (ZEPPELIN-6694): a session
+ * that has left {@code zeppelin.websocket.heartbeat.max.missed.pongs} consecutive pings
+ * unanswered is closed instead of being pinged again.
*/
void sendHeartbeat() {
+ int maxMissedPongs = zConf.getWebsocketHeartbeatMaxMissedPongs();
for (NotebookSocket conn : connectionManager.connectedSockets) {
try {
+ if (maxMissedPongs > 0 && conn.getUnansweredPings() >= maxMissedPongs) {
+ reapDeadConnection(conn, maxMissedPongs);
+ continue;
+ }
conn.sendPing();
} catch (RuntimeException e) {
LOGGER.warn("Failed to send heartbeat ping to {}", conn, e);
}
}
}
+ /**
+ * Removes a connection that stopped answering pings and closes its session. The connection is
+ * dropped from all bookkeeping before closing, because a dead peer never completes the close
+ * handshake and {@link #onClose} may therefore arrive late or not at all. Removing it from
+ * {@code sessionIdNotebookSocketMap} first also makes a later {@code onClose} a no-op.
+ */
+ private void reapDeadConnection(NotebookSocket conn, int maxMissedPongs) {
+ LOGGER.warn("Closing websocket to {}: {} consecutive heartbeat pings unanswered, last pong at {}",
+ conn, maxMissedPongs, new Date(conn.getLastPongTimestamp()));
+ String sessionId = conn.getSessionId();
+ if (sessionId != null) {
+ sessionIdNotebookSocketMap.remove(sessionId);
+ }
Review Comment:
Since `onOpen` uses `session.getId()` as a `ConcurrentHashMap` key, a registered connection can never have a null session id. This null check seems to be needed only because the mock in the test returns null from `getSessionId()`. We could stub it in the test, e.g. `when(dead.getSessionId()).thenReturn("dead-session");`, and drop the check.
```suggestion
sessionIdNotebookSocketMap.remove(conn.getSessionId());
```
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -251,6 +252,8 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
if (checkOrigin(origin)) {
NotebookSocket notebookSocket = sessionIdNotebookSocketMap
.computeIfAbsent(session.getId(), unused -> new NotebookSocket(session, headers));
+ // Per-session pong handler so the heartbeat can tell live peers from dead ones.
+ session.addMessageHandler(PongMessage.class, pong -> notebookSocket.onPong());
Review Comment:
All existing tests still pass with this line removed. The socket tests call the `onOpen(NotebookSocket)` overload directly, so none of them go through `onOpen(Session, EndpointConfig)`, where this handler is registered. Without this line, every healthy client gets disconnected after a few heartbeats (I checked this locally), so it would be nice to have a test covering this path.
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -319,19 +322,45 @@ synchronized void stopHeartbeatScheduler() {
* point of this heartbeat: it keeps connections alive even when the client-side application
* keep-alive timer is throttled or stopped (e.g. a backgrounded browser tab). A single
* session failing to receive a ping must not stop the remaining sessions from being pinged.
- * Pong responses are not tracked; once the heartbeat is enabled, Jetty's idle timeout no
- * longer determines connection liveness (see ZEPPELIN-6694).
+ *
+ * <p>Because these writes keep resetting Jetty's idle timer, the idle timeout can no longer
+ * detect dead clients. Liveness is therefore tracked explicitly (ZEPPELIN-6694): a session
+ * that has left {@code zeppelin.websocket.heartbeat.max.missed.pongs} consecutive pings
+ * unanswered is closed instead of being pinged again.
*/
void sendHeartbeat() {
+ int maxMissedPongs = zConf.getWebsocketHeartbeatMaxMissedPongs();
for (NotebookSocket conn : connectionManager.connectedSockets) {
try {
+ if (maxMissedPongs > 0 && conn.getUnansweredPings() >= maxMissedPongs) {
Review Comment:
On Jetty 11.0.24, the next frame on a connection is read only after `onMessage` returns (`StringMessageSink.accept()` calls `demand(1)` after invoking the handler). Since `onMessage` runs synchronously on the Jetty thread, pongs the client sends on time are not processed while a long-running op such as `RELOAD_NOTES_FROM_REPO`, `CHECKPOINT_NOTE` or `IMPORT_NOTE` is being handled. As a result, a live connection can be reaped.
I reproduced this locally with a 1s interval, `max.missed.pongs` set to 3, and `reloadAllNotes()` taking 6s. The client kept answering pings, but the connection was closed with 1001 (`No pong received for 3 consecutive pings`) in the middle of the op, and the op's response never reached the client. With the op taking 2s instead, the counter went up to 2 and dropped back to 0 right after the op finished, even though no new ping had been sent. It looks like the queued pongs were processed at that point.
With the defaults, a single op would have to take 3-4 minutes or more, so this is rare, but the threshold goes down as the interval is lowered. How about skipping reaping for a session that is currently handling a message? Roughly:
```java
// NotebookSocket
private volatile boolean handlingMessage;
// NotebookServer#onMessage(Session, String)
conn.setHandlingMessage(true);
try {
onMessage(conn, msg);
} finally {
conn.setHandlingMessage(false);
}
// NotebookServer#sendHeartbeat
if (maxMissedPongs > 0 && !conn.isHandlingMessage()
&& conn.getUnansweredPings() >= maxMissedPongs) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ZEPPELIN-6694] Reap dead websocket connections via pong tracking [zeppelin]
Posted by "tbonelee (via GitHub)" <gi...@apache.org>.
tbonelee commented on PR #5500:
URL: https://github.com/apache/zeppelin/pull/5500#issuecomment-5853943541
Merged into master
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@zeppelin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org