feat(connector): support WebSocket source connector

[audev482]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Summary

This PR adds a new websocket source connector that lets RisingWave ingest messages directly from a WebSocket server (e.g. an IoT sensor telemetry feed, an event stream, or a subscribe-style push API), in both plain ws:// and TLS wss:// modes.

How it works

• A websocket source/table is configured with a url (ws://... or wss://..., scheme validated at creation time) plus optional connection options. It has exactly one split: a WebSocket connection is a single ordered, non-partitionable, non-replayable message stream.
• The reader connects with tokio-tungstenite (TLS via rustls + native root CAs), streams text/binary frames, skips empty keep-alive frames, and feeds each message through the standard parsing pipeline (format plain, encode json / protobuf / bytes).
• Reconnection: WebSocket servers routinely drop long-lived connections, so the reader transparently reconnects with a capped exponential backoff (1s up to 60s) instead of erroring out. The optional init.message (e.g. a subscribe request) is re-sent on every (re)connect.
• Keep-alive: ping.interval.secs (default 30, set to 0 to disable) sends WebSocket pings; pong handling is done by tungstenite.
• Stale-subscription guard: idle.timeout.secs forces a reconnect (and re-subscribe) when no data message arrives for the given period. This covers servers that silently expire subscriptions while keeping the transport (and ping/pong) alive. Disabled by default.
• Offset semantics: because a WebSocket stream cannot be replayed, the message offset is a per-reader sequence counter used for observability only (exposed via INCLUDE offset); it is not persisted, and on recovery the reader reconnects and consumes from that point on.

New connector options

option	required	default	description
url	yes	–	server URL, must be ws:// or wss://
init.message	no	–	text sent right after (re)connect, e.g. subscribe request
headers	no	–	JSON object of string pairs added to the handshake, e.g. {"Authorization": "Bearer ..."}. Enforced as a SECRET on RisingWave Cloud (enforce_secret).
ping.interval.secs	no	30	keep-alive ping interval; 0 disables
idle.timeout.secs	no	–	reconnect if no data message arrives in this many seconds

Tests

• Unit tests: connector properties parsing/validation, split JSON round-trip, and reader behavior against a real in-process server — text/binary ingestion and empty-frame skipping, init.message re-sent across reconnects, keep-alive pings, idle-timeout reconnect + re-subscribe, custom handshake headers.
• End-to-end: e2e_test/source_inline/websocket/websocket_source.slt.serial runs against a local ws_server.py fixture, covering basic ingestion, creation-time validation, bearer-auth headers, and INCLUDE offset.

Limitations

• WebSocket is not replayable: messages the server sends while the connection is down are lost (inherent to the protocol; no offset-based recovery).
• A single split — no partitioning.
• TLS uses system-trusted native root CAs; no custom CA/client-cert options in this PR.

Closes issue #21555

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• I have checked the Release Timeline and Currently Supported Versions to determine which release branches I need to cherry-pick this PR into.

Documentation

• My PR needs documentation updates.

Release note

RisingWave now supports ingesting data from WebSocket servers via a new websocket source connector. For example, IoT sensor telemetry can be consumed directly from a device gateway:

create table sensor_telemetry (
  device_id varchar,
  temperature decimal,
  humidity decimal
) with (
  connector = 'websocket',
  url = 'wss://iot.example.com/devices/stream',
  init.message = '{"op": "subscribe", "topic": "telemetry"}',
  headers = '{"Authorization": "Bearer <token>"}'
) format plain encode json;

The connector supports plain ws:// and TLS wss:// URLs, custom handshake headers, keep-alive pings, automatic reconnection with re-subscription, and an optional idle timeout to recover from silently-expired subscriptions. Note that WebSocket streams cannot be replayed, so messages sent while the connection is down are lost.

[hzxa21]

Thanks for adding WebSocket as a source connector. One context worth documenting: after this PR, we now have two WebSocket ingestion directions in the codebase, and they serve different use cases rather than replacing each other.

• Push into RW: implemented by feat(frontend): add WebSocket ingest with async acks #25444 as WebSocket ingest for connector='webhook' tables:
  ws://<frontend-host>:4560/ingest/<database>/<schema>/<table>.
  In this mode, the upstream application owns the connection and pushes batched upsert/delete messages into a specific RW table. This is useful when the upstream system can actively send events to RW, similar to a webhook / ingest API.

• Pull from WebSocket: this PR adds connector='websocket', where RW connects out to an external WebSocket server and subscribes to a feed. This is useful when the upstream/provider already exposes a WebSocket feed and RW needs to consume it as a source connector.

So I think the product positioning should be:
#25444 = RW as WebSocket ingest endpoint
#26546 = RW as WebSocket subscriber/source connector

It would be helpful to mention this distinction the PR description as well as in the release notes, so users do not assume these two WebSocket features are competing implementations of the same thing.