[ZEPPELIN-6694] Reap dead websocket connections via pong tracking

[JangAyeon]

What is this PR for?

Since ZEPPELIN-6092, heartbeat pings keep resetting Jetty's idle timer, so the idle timeout no longer detects dead clients. A dead session stays open until TCP retransmission gives up.

This PR tracks pongs to detect dead sessions explicitly:

• Register a per-session PongMessage handler. A pong resets the session's ping count.
• In sendHeartbeat, close a session after N consecutive missed pongs (GOING_AWAY, 1001). It is removed from bookkeeping first, since a dead peer may never trigger onClose.
• Skip sessions that are handling a message. Jetty reads pongs only after onMessage returns, so a long-running op would otherwise look like missed pongs.
• Add zeppelin.websocket.heartbeat.max.missed.pongs (default 3, <= 0 disables). With the default 60s interval, a dead client is reaped in ~3-4 minutes.

What type of PR is it?

Improvement

Todos

• Track pongs and close sessions that miss N in a row
• Skip sessions that are handling a message
• Add config, docs and zeppelin-site.xml.template
• Update docs and zeppelin-site.xml.template
• Add unit tests

What is the Jira issue?

ZEPPELIN-6694

How should this be tested?

• Unit tests added:
  • NotebookSocketTest: 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, while a healthy session keeps receiving pings
    • nothing is closed when reaping is disabled
    • a session that is handling a message is not reaped
    • onMessage marks the session as handling a message until it returns
    • onOpen(Session, EndpointConfig) registers a pong handler that resets the ping count
  • 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

[tbonelee]

We could remove unnecessary comment.

Suggested change

// Per-session pong handler so the heartbeat can tell live peers from dead ones.

[tbonelee]

We could let the method name carry this and drop the comment. (sendHeartbeat() and the tests would need the same rename.)

Suggested change

	/** Number of pings sent since the last pong was received. */
	public int getUnansweredPings() {
	public int getPingsSinceLastPong() {

[tbonelee]

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.

Suggested change

	// 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.
	// Consecutive unanswered pings before a session is closed as dead. <= 0 disables reaping.

[tbonelee]

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:

// 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) {

[tbonelee]

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.

Suggested change

	String sessionId = conn.getSessionId();
	if (sessionId != null) {
	sessionIdNotebookSocketMap.remove(sessionId);
	}
	sessionIdNotebookSocketMap.remove(conn.getSessionId());

[tbonelee]

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.

[JangAyeon]

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.