zhangshenghang opened a new issue, #12713: URL: https://github.com/apache/seatunnel/issues/12713
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened `WebSocketSourceClient` records connection failures in a `fatalError` field, which the reader rethrows on every `pollNext()` via `rethrowFatalIfNeeded()`. This field is **never cleared** — not even when a reconnect succeeds. `onOpen()` resets the reconnect counter but leaves any previously recorded error in place: ```java // WebSocketSourceClient.java public void onOpen(WebSocket webSocket, Response response) { reconnectTimes.set(0); // counter is reset ... ... for (int i = 0; i < openMessages.size(); i++) { if (!webSocket.send(openMessages.get(i))) { // ... but fatalError from the previous (dead) connection survives fatalError = new WebSocketConnectorException(SEND_MESSAGE_FAILED, ...); return; } } } ``` A grep of `WebSocketSourceClient.java` confirms `fatalError` is only ever **assigned** (lines 180, 188, 203, 256) and read (line 124); it is never set back to `null` anywhere. Deterministic failure sequence with default options (`enable_reconnect = true`, `poll_timeout_ms = 1000`, `reconnect_interval_ms = 3000`): 1. `open_messages` is configured (the documented usage for subscription/auth payloads). 2. The server drops the connection right after the handshake (LB idle reset, auth race, ...). `webSocket.send()` returns `false` while sending the open messages, so `fatalError` is set to `SEND_MESSAGE_FAILED`. 3. The socket terminates, `onClosed`/`onFailure` → `onDisconnected()` schedules a reconnect — exactly what `enable_reconnect` is for. 4. The reconnect **succeeds**: the new `onOpen` re-sends the open messages and data flows again. 5. The next `pollNext()` calls `rethrowFatalIfNeeded()`, which throws the **stale** `SEND_MESSAGE_FAILED` recorded by the dead connection → the fully recovered task is killed. Additionally, with the default `poll_timeout_ms` (1000 ms) being smaller than `reconnect_interval_ms` (3000 ms), the reader hits the stale error before the first reconnect attempt even fires. So whenever `open_messages` is configured, the default auto-reconnect cannot recover the task from any transient disconnect. Expected behavior: a successful (re)connection invalidates errors recorded for the previous connection, i.e. `onOpen` should clear `fatalError`. ### SeaTunnel Version dev (commit 0278a6a74); the WebSocket source was introduced in #12465 ### SeaTunnel Config ```conf env { job.mode = "STREAMING" } source { WebSocket { url = "wss://example.com/stream" open_messages = [ """{"op":"subscribe","args":["trade.BTCUSDT"]}""" ] format = json schema = { fields { symbol = "STRING" price = "DOUBLE" } } } } sink { Console {} } ``` ### Running Command ```shell $SEATUNNEL_HOME/bin/seatunnel.sh --config ./websocket.conf -e run ``` ### Error Exception ```log org.apache.seatunnel.connectors.seatunnel.websocket.exception.WebSocketConnectorException: ErrorCode:[WEBSOCKET-xx], ErrorDescription:[Failed to send open message ...] - Failed to send open message at index [0] to websocket server [wss://example.com/stream] ``` …thrown by a task whose connection had already been re-established and was delivering data again. ### Zeta or Flink or Spark Version Zeta ### Java or Scala Version Java 17 ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
