You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@zeppelin.apache.org by "HwangRock (via GitHub)" <gi...@apache.org> on 2026/08/21 16:45:07 UTC
[PR] [ZEPPELIN-6092] Send server-initiated websocket ping frames to keep connections alive [zeppelin]
HwangRock opened a new pull request, #5432:
URL: https://github.com/apache/zeppelin/pull/5432
### What is this PR for?
Zeppelin's websocket keep-alive is client-driven only. Both UIs send an application-level `{"op":"PING"}` every 10 seconds (`websocket-event.factory.js`, `message.ts`) and the server answers nothing — `case PING:` in `NotebookServer.onMessage()` is a bare `break`. The server never writes first, so a connection survives only as long as the client's timer keeps firing.
That timer is not reliable. Chrome's intensive throttling drops background-tab timers to once per minute once the tab has been hidden for a few seconds, the chain count reaches 5, and no WebRTC is in use — all of which a 10-second `setInterval` satisfies within a minute. An open websocket is not an exemption; only WebRTC is. A discarded tab or a sleeping laptop stops the timer outright. When it stops, nothing resets the idle timer and the connection dies.
This PR has the server send a WebSocket protocol ping frame on a schedule. Per RFC 6455 section 5.5.2 the peer answers with a pong automatically, so no client-side change is required, and writing to the session resets Jetty's idle timeout — `SocketChannelEndPoint.flush()` calls `IdleTimeout.notIdle()` — along with any intermediate proxy's idle timer. This is what the nginx websocket proxying guide recommends as well: "the proxied server can be configured to periodically send WebSocket ping frames to reset the timeout and check if the connection is still alive."
The PR also makes the idle timeout itself configurable. `setupNotebookServer()` in `ZeppelinServer` sets the text message buffer size but never calls `setDefaultMaxSessionIdleTimeout()`, so the effective value has always been whatever Jetty defaults to. That default is not stable across versions: `WebSocketPolicy` used 300000ms under Jetty 9, while `WebSocketConstants.DEFAULT_IDLE_TIMEOUT` is 30 seconds under Jetty 11. The new key restores an explicit value and gives operators a knob.
Two new keys, documented in `zeppelin-site.xml.template`, `zeppelin-env.sh.template`, and `docs/setup/operation/configuration.md`:
| Key | Default | Meaning |
|---|---|---|
| `zeppelin.websocket.heartbeat.interval` | 60000 | Interval in ms between server-initiated ping frames. `0` or negative disables the heartbeat. |
| `zeppelin.websocket.idle.timeout` | 300000 | Idle timeout in ms applied to websocket sessions. |
`NotebookSocket.sendPing()` swallows and logs its exceptions so one dead session cannot break the loop over the others. The scheduler runs on a single daemon thread and starts lazily on the first connection.
### What type of PR is it?
Improvement
### What is the Jira issue?
[ZEPPELIN-6092](https://issues.apache.org/jira/browse/ZEPPELIN-6092)
### How should this be tested?
Unit tests cover the ping send path, the disabled-when-non-positive case, isolation of a failing session, and the two configuration keys.
End to end, with a client that sends nothing after the handshake, against Jetty 11's 30-second default idle timeout:
Before
```
[+0000.0s] connected (HTTP/1.1 101 Switching Protocols)
[+0030.0s] closed code=1001 reason='Connection Idle Timeout'
[+0030.0s] RESULT [A-direct]: closed code=1001 reason='Connection Idle Timeout'
```
```
ERROR NotebookServer.java[onError] - Error in WebSocket Session to /[0:0:0:0:0:0:0:1]:59302
org.eclipse.jetty.websocket.core.exception.WebSocketTimeoutException: Connection Idle Timeout
at org.eclipse.jetty.websocket.core.internal.WebSocketConnection.onIdleExpired(WebSocketConnection.java:242)
at org.eclipse.jetty.io.IdleTimeout.checkIdleTimeout(IdleTimeout.java:170)
```
After — same client, same 30-second idle timeout, heartbeat pinned to 10 seconds
```
[+0000.0s] connected (HTTP/1.1 101 Switching Protocols)
[+0120.0s] RESULT [A-direct]: still connected after 120s (no disconnect)
```
```
INFO NotebookServer.java[startHeartbeatScheduler] - Started websocket heartbeat scheduler with interval 10000 ms
```
Zero `WebSocketTimeoutException` in the server log across the 120-second window. The idle timeout is identical between the two runs, so the difference comes from the heartbeat writes alone.
To reproduce: set `zeppelin.websocket.idle.timeout` to `30000` and `zeppelin.websocket.heartbeat.interval` to `10000`, open a websocket to `/ws`, send nothing after the handshake, and watch for a close past the 30-second mark.
### Questions:
* Does the license files need updating? No
* Are there breaking changes for older versions? No. The heartbeat uses protocol ping frames, which every websocket client answers automatically. No message op was added or changed, and the existing client-driven `PING` path is untouched.
* Does this need documentation? Yes — both keys are documented in the config template, the env template, and the configuration docs.
--
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-6092] Send server-initiated websocket ping frames to keep connections alive [zeppelin]
Posted by "tbonelee (via GitHub)" <gi...@apache.org>.
tbonelee commented on code in PR #5432:
URL: https://github.com/apache/zeppelin/pull/5432#discussion_r3889337354
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -265,6 +272,73 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
public void onOpen(NotebookSocket conn) {
connectionManager.addConnection(conn);
+ startHeartbeatScheduler();
+ }
+
+ /**
+ * Starts the websocket heartbeat scheduler on first use. Idempotent: a second call while the
+ * scheduler is already running is a no-op. zConf and connectionManager are both required and
+ * are set via setter injection before any real connection can open, so starting lazily here
+ * (rather than from the injected setters, whose call order is not guaranteed) is safe.
+ */
+ synchronized void startHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ return;
+ }
+ long intervalMs = zConf.getWebsocketHeartbeatInterval();
+ if (intervalMs <= 0) {
+ LOGGER.info("Websocket heartbeat is disabled (zeppelin.websocket.heartbeat.interval={})", intervalMs);
+ return;
+ }
+ heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread thread = new Thread(r, "NotebookServer-Heartbeat");
+ thread.setDaemon(true);
+ return thread;
+ });
+ heartbeatScheduler.scheduleAtFixedRate(
+ this::sendHeartbeat, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+ heartbeatShutdownHook = new Thread(this::stopHeartbeatScheduler);
+ Runtime.getRuntime().addShutdownHook(heartbeatShutdownHook);
+ LOGGER.info("Started websocket heartbeat scheduler with interval {} ms", intervalMs);
+ }
+
+ /**
+ * Stops the websocket heartbeat scheduler, if running, and deregisters its shutdown hook so
+ * repeated start/stop cycles do not accumulate hooks. Safe to call multiple times and safe
+ * to call when the scheduler was never started.
+ */
+ synchronized void stopHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ heartbeatScheduler.shutdownNow();
+ heartbeatScheduler = null;
+ }
+ if (heartbeatShutdownHook != null && Thread.currentThread() != heartbeatShutdownHook) {
+ try {
+ Runtime.getRuntime().removeShutdownHook(heartbeatShutdownHook);
+ } catch (IllegalStateException e) {
+ // JVM is already shutting down; the hook will simply run (as a harmless no-op).
+ }
+ heartbeatShutdownHook = null;
+ }
+ }
+
+ /**
+ * Sends a WebSocket protocol ping frame to every connected session. Writing to a session
+ * resets Jetty's idle timeout (and any intermediate proxy's idle timer), which is the whole
+ * 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.
+ */
+ void sendHeartbeat() {
+ synchronized (connectionManager.connectedSockets) {
Review Comment:
Yes, we could address that in a separate PR.
--
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-6092] Send server-initiated websocket ping frames to keep connections alive [zeppelin]
Posted by "HwangRock (via GitHub)" <gi...@apache.org>.
HwangRock commented on code in PR #5432:
URL: https://github.com/apache/zeppelin/pull/5432#discussion_r3889244729
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -265,6 +272,73 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
public void onOpen(NotebookSocket conn) {
connectionManager.addConnection(conn);
+ startHeartbeatScheduler();
+ }
+
+ /**
+ * Starts the websocket heartbeat scheduler on first use. Idempotent: a second call while the
+ * scheduler is already running is a no-op. zConf and connectionManager are both required and
+ * are set via setter injection before any real connection can open, so starting lazily here
+ * (rather than from the injected setters, whose call order is not guaranteed) is safe.
+ */
+ synchronized void startHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ return;
+ }
+ long intervalMs = zConf.getWebsocketHeartbeatInterval();
+ if (intervalMs <= 0) {
Review Comment:
You're right. I'll apply this as suggested, resetting the flag in `stopHeartbeatScheduler()`.
--
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-6092] Send server-initiated websocket ping frames to keep connections alive [zeppelin]
Posted by "tbonelee (via GitHub)" <gi...@apache.org>.
tbonelee commented on PR #5432:
URL: https://github.com/apache/zeppelin/pull/5432#issuecomment-5541926821
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
Re: [PR] [ZEPPELIN-6092] Send server-initiated websocket ping frames to keep connections alive [zeppelin]
Posted by "tbonelee (via GitHub)" <gi...@apache.org>.
tbonelee merged PR #5432:
URL: https://github.com/apache/zeppelin/pull/5432
--
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-6092] Send server-initiated websocket ping frames to keep connections alive [zeppelin]
Posted by "HwangRock (via GitHub)" <gi...@apache.org>.
HwangRock commented on code in PR #5432:
URL: https://github.com/apache/zeppelin/pull/5432#discussion_r3889234729
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -265,6 +272,73 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
public void onOpen(NotebookSocket conn) {
connectionManager.addConnection(conn);
+ startHeartbeatScheduler();
+ }
+
+ /**
+ * Starts the websocket heartbeat scheduler on first use. Idempotent: a second call while the
+ * scheduler is already running is a no-op. zConf and connectionManager are both required and
+ * are set via setter injection before any real connection can open, so starting lazily here
+ * (rather than from the injected setters, whose call order is not guaranteed) is safe.
+ */
+ synchronized void startHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ return;
+ }
+ long intervalMs = zConf.getWebsocketHeartbeatInterval();
+ if (intervalMs <= 0) {
+ LOGGER.info("Websocket heartbeat is disabled (zeppelin.websocket.heartbeat.interval={})", intervalMs);
+ return;
+ }
+ heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread thread = new Thread(r, "NotebookServer-Heartbeat");
+ thread.setDaemon(true);
+ return thread;
+ });
+ heartbeatScheduler.scheduleAtFixedRate(
+ this::sendHeartbeat, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+ heartbeatShutdownHook = new Thread(this::stopHeartbeatScheduler);
+ Runtime.getRuntime().addShutdownHook(heartbeatShutdownHook);
+ LOGGER.info("Started websocket heartbeat scheduler with interval {} ms", intervalMs);
+ }
+
+ /**
+ * Stops the websocket heartbeat scheduler, if running, and deregisters its shutdown hook so
+ * repeated start/stop cycles do not accumulate hooks. Safe to call multiple times and safe
+ * to call when the scheduler was never started.
+ */
+ synchronized void stopHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ heartbeatScheduler.shutdownNow();
+ heartbeatScheduler = null;
+ }
+ if (heartbeatShutdownHook != null && Thread.currentThread() != heartbeatShutdownHook) {
+ try {
+ Runtime.getRuntime().removeShutdownHook(heartbeatShutdownHook);
+ } catch (IllegalStateException e) {
+ // JVM is already shutting down; the hook will simply run (as a harmless no-op).
+ }
+ heartbeatShutdownHook = null;
+ }
+ }
+
+ /**
+ * Sends a WebSocket protocol ping frame to every connected session. Writing to a session
+ * resets Jetty's idle timeout (and any intermediate proxy's idle timer), which is the whole
+ * 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.
+ */
+ void sendHeartbeat() {
+ synchronized (connectionManager.connectedSockets) {
Review Comment:
You're right. I mirrored this from the `broadcast()` methods and pulled the `synchronized` along with it, but the heartbeat doesn't need it. `connectedSockets` is a `ConcurrentLinkedQueue`, so iteration is already safe, and `addConnection`/`removeConnection` don't take this monitor either. I'll drop it so a slow `sendPing()` stays contained to the heartbeat thread.
The same is true of the `broadcast()` and `broadcastToAllConnectionsExcept()` locks, which don't buy anything on a `ConcurrentLinkedQueue` either. That's outside this issue's scope, so I'll leave them here. Would it make sense to file a follow-up ticket to clean those up as well?
--
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-6092] Send server-initiated websocket ping frames to keep connections alive [zeppelin]
Posted by "tbonelee (via GitHub)" <gi...@apache.org>.
tbonelee commented on code in PR #5432:
URL: https://github.com/apache/zeppelin/pull/5432#discussion_r3888853926
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -265,6 +272,73 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
public void onOpen(NotebookSocket conn) {
connectionManager.addConnection(conn);
+ startHeartbeatScheduler();
+ }
+
+ /**
+ * Starts the websocket heartbeat scheduler on first use. Idempotent: a second call while the
+ * scheduler is already running is a no-op. zConf and connectionManager are both required and
+ * are set via setter injection before any real connection can open, so starting lazily here
+ * (rather than from the injected setters, whose call order is not guaranteed) is safe.
+ */
+ synchronized void startHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ return;
+ }
+ long intervalMs = zConf.getWebsocketHeartbeatInterval();
+ if (intervalMs <= 0) {
+ LOGGER.info("Websocket heartbeat is disabled (zeppelin.websocket.heartbeat.interval={})", intervalMs);
+ return;
+ }
+ heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread thread = new Thread(r, "NotebookServer-Heartbeat");
+ thread.setDaemon(true);
+ return thread;
+ });
+ heartbeatScheduler.scheduleAtFixedRate(
+ this::sendHeartbeat, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+ heartbeatShutdownHook = new Thread(this::stopHeartbeatScheduler);
+ Runtime.getRuntime().addShutdownHook(heartbeatShutdownHook);
+ LOGGER.info("Started websocket heartbeat scheduler with interval {} ms", intervalMs);
+ }
+
+ /**
+ * Stops the websocket heartbeat scheduler, if running, and deregisters its shutdown hook so
+ * repeated start/stop cycles do not accumulate hooks. Safe to call multiple times and safe
+ * to call when the scheduler was never started.
+ */
+ synchronized void stopHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ heartbeatScheduler.shutdownNow();
+ heartbeatScheduler = null;
+ }
+ if (heartbeatShutdownHook != null && Thread.currentThread() != heartbeatShutdownHook) {
+ try {
+ Runtime.getRuntime().removeShutdownHook(heartbeatShutdownHook);
+ } catch (IllegalStateException e) {
+ // JVM is already shutting down; the hook will simply run (as a harmless no-op).
+ }
+ heartbeatShutdownHook = null;
+ }
+ }
+
+ /**
+ * Sends a WebSocket protocol ping frame to every connected session. Writing to a session
+ * resets Jetty's idle timeout (and any intermediate proxy's idle timer), which is the whole
Review Comment:
With the heartbeat in place, it looks like the server-side write alone keeps resetting Jetty's idle timer (`SocketChannelEndPoint.flush()` calls `notIdle()` whenever `flushed > 0`). Since we don't check whether a pong comes back, the "reap dead connections" role that the idle timeout was implicitly playing gets weaker, and session cleanup may end up deferred until the OS gives up on TCP retransmission.
I don't think this is a reason to block the PR. The old reaping behavior couldn't tell a backgrounded tab from a dead client, which is the very problem this issue starts from. It would just be nice to have the trade-off recorded in the code for whoever reads it later. Would it make sense to add a line to this javadoc saying that pongs are not tracked and the idle timeout no longer determines liveness, and to split pong-based cleanup (a per-session `PongMessage` handler plus closing after N consecutive misses) into a follow-up issue? Please let me know if I'm missing something here.
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -265,6 +272,73 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
public void onOpen(NotebookSocket conn) {
connectionManager.addConnection(conn);
+ startHeartbeatScheduler();
+ }
+
+ /**
+ * Starts the websocket heartbeat scheduler on first use. Idempotent: a second call while the
+ * scheduler is already running is a no-op. zConf and connectionManager are both required and
+ * are set via setter injection before any real connection can open, so starting lazily here
+ * (rather than from the injected setters, whose call order is not guaranteed) is safe.
+ */
+ synchronized void startHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ return;
+ }
+ long intervalMs = zConf.getWebsocketHeartbeatInterval();
+ if (intervalMs <= 0) {
+ LOGGER.info("Websocket heartbeat is disabled (zeppelin.websocket.heartbeat.interval={})", intervalMs);
+ return;
+ }
+ heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread thread = new Thread(r, "NotebookServer-Heartbeat");
+ thread.setDaemon(true);
+ return thread;
+ });
+ heartbeatScheduler.scheduleAtFixedRate(
+ this::sendHeartbeat, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+ heartbeatShutdownHook = new Thread(this::stopHeartbeatScheduler);
Review Comment:
I read this as `NotebookServer` wanting to manage its own lifecycle, and I think that direction is natural in itself. Whatever starts the scheduler should be what stops it. It just seems a little awkward to do that self-contained given the current structure.
- HK2's `@PreDestroy` would be the natural place, but `ZeppelinServer` never calls `sharedServiceLocator.shutdown()`, so the callback doesn't appear to run. (`SearchService` and `LuceneSearch` already carry `@PreDestroy` and seem to be in the same never-invoked state.)
- Stopping it from `@OnClose` when the last connection goes away is another option, but with connection churn it would repeatedly start and stop, which has its own downside.
Also, a JVM shutdown hook seems to cover only part of the normal shutdown. `ZeppelinServer` registers its own hook at `ZeppelinServer.java:251`, so on the SIGTERM path the two hooks end up running in parallel within the same event. But `shutdown()` only calls `System.exit()` when the exit code is non-zero, so in cases like `MiniZeppelinServer.shutDown()`, which calls `zepServer.close()` while the JVM stays alive, the hook doesn't run. Starting and stopping the server repeatedly in tests could then accumulate a scheduler and a hook per instance.
So in practice it seems like some form of `ZeppelinServer` knowing about this is what remains. Conveniently, `NotebookServer` is bound as an HK2 singleton (`ZeppelinServer.java:202`) and `SessionConfigurator.getEndpointInstance()` pulls that same instance, so this could follow exactly the pattern used for `Notebook.close()` right next to it:
```java
if (sharedServiceLocator != null) {
sharedServiceLocator.getService(NotebookServer.class).stopHeartbeatScheduler();
...
```
With that, the existing hook covers the JVM kill path, so the dedicated hook added here along with the `removeShutdownHook` / `IllegalStateException` handling and the self-reference guard could probably be dropped. The thread is a daemon anyway.
Unifying the shutdown path around `@PreDestroy` + `locator.shutdown()`, so that `ZeppelinServer` doesn't need to know about individual services, feels like the better picture to me. But things like the conditional shutdown in recovery mode are tangled up in it, so touching that in this PR seems like a lot to ask. Would it make sense to leave it as a separate issue?
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -265,6 +272,73 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
public void onOpen(NotebookSocket conn) {
connectionManager.addConnection(conn);
+ startHeartbeatScheduler();
+ }
+
+ /**
+ * Starts the websocket heartbeat scheduler on first use. Idempotent: a second call while the
+ * scheduler is already running is a no-op. zConf and connectionManager are both required and
+ * are set via setter injection before any real connection can open, so starting lazily here
+ * (rather than from the injected setters, whose call order is not guaranteed) is safe.
+ */
+ synchronized void startHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ return;
+ }
+ long intervalMs = zConf.getWebsocketHeartbeatInterval();
+ if (intervalMs <= 0) {
Review Comment:
When `interval <= 0`, `heartbeatScheduler` stays null, so it looks like this method re-runs in full on every connection and logs "Websocket heartbeat is disabled" each time.
`ZeppelinConfiguration` seems to be loaded once at startup with no runtime mutation path (`ConfigurationsRestApi` is GET-only), so evaluating this once should be enough:
```java
private boolean heartbeatInitialized;
synchronized void startHeartbeatScheduler() {
if (heartbeatInitialized) {
return;
}
heartbeatInitialized = true;
...
}
```
Resetting `heartbeatInitialized = false` in `stopHeartbeatScheduler()` should keep the restart scenario in the tests working as is.
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -265,6 +272,73 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
public void onOpen(NotebookSocket conn) {
connectionManager.addConnection(conn);
+ startHeartbeatScheduler();
+ }
+
+ /**
+ * Starts the websocket heartbeat scheduler on first use. Idempotent: a second call while the
+ * scheduler is already running is a no-op. zConf and connectionManager are both required and
+ * are set via setter injection before any real connection can open, so starting lazily here
+ * (rather than from the injected setters, whose call order is not guaranteed) is safe.
+ */
+ synchronized void startHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ return;
+ }
+ long intervalMs = zConf.getWebsocketHeartbeatInterval();
+ if (intervalMs <= 0) {
+ LOGGER.info("Websocket heartbeat is disabled (zeppelin.websocket.heartbeat.interval={})", intervalMs);
+ return;
+ }
+ heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread thread = new Thread(r, "NotebookServer-Heartbeat");
+ thread.setDaemon(true);
+ return thread;
+ });
+ heartbeatScheduler.scheduleAtFixedRate(
+ this::sendHeartbeat, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+ heartbeatShutdownHook = new Thread(this::stopHeartbeatScheduler);
+ Runtime.getRuntime().addShutdownHook(heartbeatShutdownHook);
+ LOGGER.info("Started websocket heartbeat scheduler with interval {} ms", intervalMs);
+ }
+
+ /**
+ * Stops the websocket heartbeat scheduler, if running, and deregisters its shutdown hook so
+ * repeated start/stop cycles do not accumulate hooks. Safe to call multiple times and safe
+ * to call when the scheduler was never started.
+ */
+ synchronized void stopHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ heartbeatScheduler.shutdownNow();
+ heartbeatScheduler = null;
+ }
+ if (heartbeatShutdownHook != null && Thread.currentThread() != heartbeatShutdownHook) {
+ try {
+ Runtime.getRuntime().removeShutdownHook(heartbeatShutdownHook);
+ } catch (IllegalStateException e) {
+ // JVM is already shutting down; the hook will simply run (as a harmless no-op).
+ }
+ heartbeatShutdownHook = null;
+ }
+ }
+
+ /**
+ * Sends a WebSocket protocol ping frame to every connected session. Writing to a session
+ * resets Jetty's idle timeout (and any intermediate proxy's idle timer), which is the whole
+ * 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.
+ */
+ void sendHeartbeat() {
+ synchronized (connectionManager.connectedSockets) {
Review Comment:
`connectedSockets` is a `ConcurrentLinkedQueue` (`ConnectionManager.java:71`), so iterating it shouldn't need the lock. Holding `synchronized` here makes the heartbeat serialize against the `broadcast()` methods on the same monitor. Was that intentional?
What slightly concerns me is that `sendPing()` blocks. In Jetty 11, `JakartaWebSocketRemoteEndpoint.sendPing()` waits for the flush to complete via `FutureCallback.block()`, and the default write timeout appears to be `WebSocketConstants.DEFAULT_WRITE_TIMEOUT = Duration.ZERO`. So it seems possible for one slow session to tie up the heartbeat thread, and if it holds the lock at the same time, broadcasts would wait alongside it. I haven't reproduced this, so I'm not sure how long it actually blocks in practice.
Dropping just the `synchronized` block would probably keep the impact contained to the heartbeat thread:
```java
void sendHeartbeat() {
for (NotebookSocket conn : connectionManager.connectedSockets) {
try {
conn.sendPing();
} catch (RuntimeException e) {
LOGGER.warn("Failed to send heartbeat ping to {}", conn, e);
}
}
}
```
There does seem to be a path for bounding the block itself, since `session.getAsyncRemote().setSendTimeout(...)` in `onOpen` maps down to `coreSession.setWriteTimeout()`, but that would change the behavior of the existing text sends too, so it feels out of scope for this PR.
--
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-6092] Send server-initiated websocket ping frames to keep connections alive [zeppelin]
Posted by "HwangRock (via GitHub)" <gi...@apache.org>.
HwangRock commented on code in PR #5432:
URL: https://github.com/apache/zeppelin/pull/5432#discussion_r3889255289
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -265,6 +272,73 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
public void onOpen(NotebookSocket conn) {
connectionManager.addConnection(conn);
+ startHeartbeatScheduler();
+ }
+
+ /**
+ * Starts the websocket heartbeat scheduler on first use. Idempotent: a second call while the
+ * scheduler is already running is a no-op. zConf and connectionManager are both required and
+ * are set via setter injection before any real connection can open, so starting lazily here
+ * (rather than from the injected setters, whose call order is not guaranteed) is safe.
+ */
+ synchronized void startHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ return;
+ }
+ long intervalMs = zConf.getWebsocketHeartbeatInterval();
+ if (intervalMs <= 0) {
+ LOGGER.info("Websocket heartbeat is disabled (zeppelin.websocket.heartbeat.interval={})", intervalMs);
+ return;
+ }
+ heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread thread = new Thread(r, "NotebookServer-Heartbeat");
+ thread.setDaemon(true);
+ return thread;
+ });
+ heartbeatScheduler.scheduleAtFixedRate(
+ this::sendHeartbeat, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+ heartbeatShutdownHook = new Thread(this::stopHeartbeatScheduler);
+ Runtime.getRuntime().addShutdownHook(heartbeatShutdownHook);
+ LOGGER.info("Started websocket heartbeat scheduler with interval {} ms", intervalMs);
+ }
+
+ /**
+ * Stops the websocket heartbeat scheduler, if running, and deregisters its shutdown hook so
+ * repeated start/stop cycles do not accumulate hooks. Safe to call multiple times and safe
+ * to call when the scheduler was never started.
+ */
+ synchronized void stopHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ heartbeatScheduler.shutdownNow();
+ heartbeatScheduler = null;
+ }
+ if (heartbeatShutdownHook != null && Thread.currentThread() != heartbeatShutdownHook) {
+ try {
+ Runtime.getRuntime().removeShutdownHook(heartbeatShutdownHook);
+ } catch (IllegalStateException e) {
+ // JVM is already shutting down; the hook will simply run (as a harmless no-op).
+ }
+ heartbeatShutdownHook = null;
+ }
+ }
+
+ /**
+ * Sends a WebSocket protocol ping frame to every connected session. Writing to a session
+ * resets Jetty's idle timeout (and any intermediate proxy's idle timer), which is the whole
Review Comment:
Yes to both. I'll add the javadoc line here noting that pongs aren't tracked and the idle timeout no longer determines liveness, and I've split the pong-based cleanup into [ZEPPELIN-6694](https://issues.apache.org/jira/browse/ZEPPELIN-6694).
--
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-6092] Send server-initiated websocket ping frames to keep connections alive [zeppelin]
Posted by "HwangRock (via GitHub)" <gi...@apache.org>.
HwangRock commented on code in PR #5432:
URL: https://github.com/apache/zeppelin/pull/5432#discussion_r3889253786
##########
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java:
##########
@@ -265,6 +272,73 @@ public void onOpen(Session session, EndpointConfig endpointConfig) throws IOExce
public void onOpen(NotebookSocket conn) {
connectionManager.addConnection(conn);
+ startHeartbeatScheduler();
+ }
+
+ /**
+ * Starts the websocket heartbeat scheduler on first use. Idempotent: a second call while the
+ * scheduler is already running is a no-op. zConf and connectionManager are both required and
+ * are set via setter injection before any real connection can open, so starting lazily here
+ * (rather than from the injected setters, whose call order is not guaranteed) is safe.
+ */
+ synchronized void startHeartbeatScheduler() {
+ if (heartbeatScheduler != null) {
+ return;
+ }
+ long intervalMs = zConf.getWebsocketHeartbeatInterval();
+ if (intervalMs <= 0) {
+ LOGGER.info("Websocket heartbeat is disabled (zeppelin.websocket.heartbeat.interval={})", intervalMs);
+ return;
+ }
+ heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread thread = new Thread(r, "NotebookServer-Heartbeat");
+ thread.setDaemon(true);
+ return thread;
+ });
+ heartbeatScheduler.scheduleAtFixedRate(
+ this::sendHeartbeat, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+ heartbeatShutdownHook = new Thread(this::stopHeartbeatScheduler);
Review Comment:
Yes, that makes sense. I've filed it as [ZEPPELIN-6693](https://issues.apache.org/jira/browse/ZEPPELIN-6693), capturing both the `ZeppelinServer`-driven stop and the `@PreDestroy` + `locator.shutdown()` end state you described.
--
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