diff --git a/README.md b/README.md index 7a4ffbc..44f2538 100644 --- a/README.md +++ b/README.md @@ -5,16 +5,21 @@ `mvn -DskipTests clean package` The QuestDB client version is a Maven property (`questdb.client.version`, default -`1.3.2`, the Maven Central release). The QWP (WebSocket) wire protocol differs -between release and master builds, so the client **must match the target server -build**. To ingest into a server built from master, build with: +`1.3.6-SNAPSHOT`). This build **requires** it: the QWP transport and the lifecycle +`switch` features used here (`switch status`, the current connection-event set) are +not stable in earlier clients. `1.3.6-SNAPSHOT` is **not on Maven Central** — install +it locally first, from the `feat/connect-timeout` branch of `java-questdb-client`: ``` -mvn -DskipTests clean package -Dquestdb.client.version=1.3.5-SNAPSHOT +# in a checkout of java-questdb-client on feat/connect-timeout: +mvn -DskipTests clean install ``` -A mismatch is not reported cleanly; it surfaces as `unknown msg_kind 0x18` or -`invalid column type code: 0x0`, or as rows silently not landing. +Then build this project normally (`mvn -DskipTests clean package`), or pin another +build with `-Dquestdb.client.version=...`. The QWP (WebSocket) wire protocol **must +match the target server build** — a mismatch is not reported cleanly; it surfaces as +`unknown msg_kind 0x18` or `invalid column type code: 0x0`, or as rows silently not +landing. ## Usage Example @@ -29,21 +34,32 @@ java -jar target/ilp_sender-1.0-SNAPSHOT.jar \ --csv ./trades20250728.csv.gz \ --timestamp-from-file false \ --retry-timeout 360000 \ +--connect-timeout-ms 3000 \ --sender-id ha_sender \ --store-forward-dir /tmp/qdb-sf \ --batch-size 10000 \ --batches-per-transaction 10 ``` +`--retry-timeout` (default `360000` ms) is the overall "keep retrying" budget and applies to +**both** transports: it drives `retryTimeoutMillis` on ILP and `reconnectMaxDurationMillis` on +QWP. That is a different timeout from `--connect-timeout-ms` (below), which bounds a single +connect attempt. + ## Transport (`--protocol`) - `--protocol qwp` (default): QWP over WebSocket. Adds store-and-forward (un-acked frames spill to disk and replay after an outage) and transactional commit. Use the server's **HTTP port** (`:9000`) in `--addrs`. -- `--protocol ilp`: the previous HTTP/ILP transport, unchanged. Ignores the QWP-only - flags below. +- `--protocol qwpudp`: QWP over **UDP**, fire-and-forget datagram ingest. Use the + server's **UDP ingest port** (`:9007` by convention) in `--addrs`. It is **ingest-only, + unauthenticated, best-effort, and single-endpoint** (no failover). See + [QWP/UDP transport](#qwpudp-transport-best-effort-single-endpoint) for the full picture + and how to avoid packet loss. +- `--protocol ilp`: the previous HTTP/ILP transport, unchanged. Ignores the + QWP/WebSocket-only flags below. -QWP-only flags: +QWP/WebSocket-only flags (`qwp`; ignored by `qwpudp` and `ilp`): - `--sender-id` (default `ha_sender`): store-and-forward key. **Must be unique per process/server**; set it explicitly when running more than one process on a host. @@ -59,8 +75,13 @@ QWP-only flags: - `--batches-per-transaction` (default `10`): an explicit `flush()` commits every `batch-size × batches-per-transaction` rows atomically per table. See [Commit cadence](#commit-cadence-freshness-vs-throughput) below. +- `--connect-timeout-ms` (default `3000`, `0` uses the OS default): upper bound on a **single** + TCP connect attempt, for both the senders (`connectTimeoutMillis`) and the probe + (`connect_timeout=`). Without it a black-holed host (route accepts, never answers) rides the + OS connect timeout (~minutes) before failing over to the next `--addrs` host; with it, failover + happens in seconds. The ILP path is left unbounded so `--protocol ilp` stays identical. - `--probe-interval-ms` (default `1000`, `0` disables): interval for the probe thread - that polls the latest ingested timestamp. See [Probe](#probe-qwp-only) below. + that polls the latest ingested timestamp. See [Probe](#probe-qwpwebsocket-only) below. - `--enterprise` (default `false`): request durable acks (`requestDurableAck`), holding spilled frames until the server confirms a durable commit. **Enterprise only**: an OSS server rejects the WebSocket upgrade, so leave it off against OSS. See @@ -70,6 +91,51 @@ QWP-only flags: Enterprise; a no-op on OSS. The ingestion path is zone-blind (it must follow the primary), so this does not affect the senders. +## QWP/UDP transport (best-effort, single endpoint) + +`--protocol qwpudp` sends rows as UDP datagrams to the server's UDP ingest port (`:9007` +by convention; put it in `--addrs`). The server must have UDP ingest enabled, and it +accepts **any** connection on that port. It is the simplest and lowest-overhead +transport, at the cost of every reliability guarantee the other two provide. + +**Ingest-only, unauthenticated.** There is no query path, so the probe is not started and +`--probe-interval-ms` is ignored. UDP rejects authentication: `--token`, `--username`, and +`--password` are ignored (with a warning). Store-and-forward, TLS, and durable ack do not +apply. + +**Single endpoint, no failover.** UDP is connectionless and unacknowledged: there is no +connection to detect as dropped and no delivery feedback, so the client cannot know a host +is down and therefore **cannot fail over**. The client enforces this by rejecting more than +one address: + +``` +only a single address (host:port) is supported for UDP transport +``` + +Pass exactly one `host:9007`. If that host is down, datagrams are silently lost. For high +availability across hosts use `qwp` (WebSocket store-and-forward) or `ilp` (HTTP retry). + +**Best-effort: datagrams can be dropped.** There is no ack, no retransmit, and no +store-and-forward, so a lost datagram is a lost row. UDP also has **no backpressure**: if +the sender emits datagrams faster than the server drains its OS receive buffer, the buffer +overflows and datagrams are dropped **wholesale**, not one at a time. + +That burst sensitivity is the main thing to tune. Rows are flushed to datagrams every +`--batch-size` rows, so a large batch flushed all at once (especially by several workers +finishing together) is exactly the burst that overflows the buffer. Observed on loopback: + +| Config | Result | +| --- | --- | +| `--num-senders 4 --batch-size 10000` (one big burst per worker) | **0 of 8000 rows landed** | +| `--num-senders 4 --batch-size 100` (paced, small flushes) | **8000 of 8000 rows landed** | + +So for UDP, **keep `--batch-size` small** (e.g. `100`-`500`) to pace the datagrams, and if +you still see loss, raise the OS UDP receive-buffer size on the server host +(`net.core.rmem_max` / `rmem_default` on Linux). Even then, treat UDP as best-effort: +enable dedup and reconcile counts if you need to know what landed. Timestamp behavior is +the same as the other transports (a single worker stamps client-side micros; multiple +workers use server-side `atNow()`; see [Timestamps](#timestamps-and-throughput)). + ## Commit cadence (freshness vs throughput) On QWP the sender runs in `transactional(true)` mode. That decouples *flushing* from @@ -97,24 +163,35 @@ seconds between commits (per worker) = (batch-size × batches-per-transaction) / for many seconds. Keep `--batches-per-transaction` small when freshness matters more than raw throughput. -## Probe (QWP only) +## Probe (QWP/WebSocket only) -When the transport is QWP, a separate thread polls the latest ingested timestamp and +When the transport is `qwp` (WebSocket), a separate thread polls the latest ingested timestamp and prints it to stdout once per `--probe-interval-ms` (default `1000` ms; `0` disables). It runs `select timestamp from trades limit -1` over a **QWP query client** and is fully independent of the senders, so it does not affect ingestion. ``` -[probe] latest trades timestamp = 2026-07-01T14:31:40.591545Z (raw=1782916300591545) +[probe] latest trades timestamp = 2026-07-01T14:31:40.591545Z (raw=1782916300591545) served by role=PRIMARY node=n1 zone=eu-west-1 ``` +The `served by role=...` is the **live** lifecycle role (`current_role` from `switch status`) +of the node currently answering the probe, run on it each poll. It is authoritative: the QWP +handshake `SERVER_INFO` role is only refreshed on a *reconnect*, so after an in-place +primary→replica switch (which does not drop the read connection) that role goes stale, +whereas `switch status` always reflects the truth, so the role flips the moment the serving +node changes role. `node`/`zone` come from the handshake `SERVER_INFO`. While a switch is in +flight, `(switching -> ROLE)` is appended (the `target_role` differs from `current_role`). If +the status query is unavailable (OSS, or the token lacks `SYSTEM ADMIN`), the probe falls back +to the handshake role, explicitly labelled `(handshake role; live 'switch status' unavailable, +may be stale)`. + The query client uses the **same host list and token/auth** as the senders and enables **failover** when more than one host is given (`ws|wss::addr=h1,h2,...;...;failover=on`), so if a host stops responding it moves to the next one automatically and narrates the transition (see [Failover and connection narration](#failover-and-connection-narration-qwp)). Before the `trades` table exists it prints `[query client] server error: table does not -exist` each interval, then switches to timestamps once ingestion begins. ILP ignores -`--probe-interval-ms`. +exist` each interval, then switches to timestamps once ingestion begins. `ilp` and +`qwpudp` ignore `--probe-interval-ms` (neither has a query path). On connect the probe prints the instance actually serving it: @@ -145,7 +222,26 @@ The **ingestion client** (one per worker) reports connection-state changes via t [ingestion client ha_sender-0] all endpoints unreachable, backing off ``` -(also `auth failed` and `reconnect budget exhausted` terminal states.) +(also an `auth failed` terminal state; any other event kind the client emits is narrated +generically by the `default` branch.) + +A failed QWP WebSocket upgrade is otherwise hard to read from the raw client error, so +wherever it surfaces (the `endpoint ... failed` line and the fatal `Sender N got error` / +`Worker failed` lines) the sender appends a hint. It uses the client's **typed** signals, not +string-matching, so it tells the causes apart: + +- **`QwpAuthFailedException` (HTTP 401/403)** — bad/expired credentials or missing permission: + `-- auth rejected (HTTP 401): bad/expired credentials or missing permission; check --token or --user/--password` +- **`WebSocketUpgradeException.isRoleMismatch()` (HTTP 421)** — the endpoint is a replica, i.e. + no primary is available among `--addrs` to accept writes: + `-- endpoint is not writable (role=REPLICA): no primary available among --addrs to accept writes` +- **`WebSocketUpgradeException` status 400** — a missing or malformed token (e.g. an unset token + env var); the server refuses the upgrade before it can return a 401: + `-- HTTP 400 on the upgrade: likely a missing or malformed ILP auth token (e.g. an unset token env var); check --token or --user/--password` + +``` +Sender 0 got error: io.questdb.client...WebSocketUpgradeException: WebSocket upgrade failed: HTTP/1.1 400 Bad request -- HTTP 400 on the upgrade: likely a missing or malformed ILP auth token (e.g. an unset token env var); check --token or --user/--password +``` The **query client** (the probe) reports, prefixed `[query client]`: @@ -186,6 +282,36 @@ timestamps, so O3 is still avoided, and this sidesteps the per-batch stamping ab (giving near-distinct per-row timestamps). ILP, or QWP with more than one worker, keep using server-side `atNow()`, which stays O3-safe across concurrent senders. +## Row id and dedup (`trade_id`) + +Every row the trades sender writes carries a `trade_id` column of the form +`-` (for example `0-1`, `0-2`, ...), where the sequence is 1-based and +monotonic per sender. It is a high-cardinality **VARCHAR**, deliberately not a symbol (each +row is unique, so a symbol column would blow up the symbol table). + +Two uses: + +- **Completeness check, timestamp-independent.** Each sender emits a contiguous `1..N`, so you + can confirm nothing was lost and spot gaps regardless of how rows were timestamped. +- **Idempotent replay via dedup.** Store-and-forward is at-least-once: on a failover the client + replays the un-acked in-flight transaction to the new primary, re-writing the rows the old + primary had already durably persisted (send 100,000 rows, kill the primary mid-transaction, + and you land ~100,101). Enabling dedup collapses those back to exactly once: + + ```sql + ALTER TABLE trades DEDUP ENABLE UPSERT KEYS(timestamp, trade_id); + ``` + + Dedup keys must include the designated timestamp, so the match is on `(timestamp, trade_id)`. + `trade_id` also keeps legitimately-distinct rows that share a microsecond apart (common at + millions of rows/s), which a `(symbol, side, timestamp)` key alone would wrongly merge. + +**Requirement:** dedup only catches replays when the row carries a **client-side timestamp**, so +the timestamp is baked into the spilled frame and reproduced identically on replay. That means a +single worker (the default `at(Instant.now())` path) or `--timestamp-from-file` / +`--seconds-offset`. Under multi-worker `atNow()` the new primary assigns a fresh timestamp on +replay and dedup cannot match. See [Timestamps and throughput](#timestamps-and-throughput). + ## Benchmarking throughput On completion the sender prints the ingestion time and rate, for example: @@ -242,22 +368,35 @@ QWP WebSocket transport. ## Primary failover watchdog `qdb-primary-watchdog.sh` is a standalone Bash watchdog for an Enterprise cluster. Give it the -ordered list of every node's lifecycle admin endpoint (`host:9003`); it finds the current -primary, health-checks it, and on failure promotes another node via the lifecycle REST API. +ordered list of every node's lifecycle admin endpoint (`host:9003`); it finds (or, on a +fresh cluster, elects) the current primary, monitors its role, and on failure promotes another +node via the lifecycle REST API. Behaviour: -- **Detect:** on startup it sweeps the list and monitors whichever node reports - `currentRole:PRIMARY`. -- **Step down first:** when the primary stops responding it first sends a best-effort - `{"role":"replica"}` to the old primary, so a false positive (it is actually still up) cannot - leave two primaries. -- **Promote in order:** it then promotes the first reachable node in list order (earliest first, - skipping the dead one), retrying with exponential backoff. With `a,b,c`, if `b` is primary and - dies it fails back to `a` when `a` is available, and only reaches `c` if `a` cannot be promoted. - It then monitors the newly promoted node, repeating indefinitely. -- **Give up:** exits only if no node can be promoted (whole cluster down), or if no primary is - found at startup. Under systemd it is then restarted and re-detects the primary. +- **Detect / bootstrap:** on startup it sweeps the list and monitors whichever node reports + `currentRole:PRIMARY`. If **no** node is primary yet (a freshly-started cluster) it promotes + the first healthy node in list order, so the cluster always ends up with a primary. If every + node is unhealthy it keeps retrying rather than exiting. Split-brain is not a concern here: + QuestDB stops the offending node if two ever claim primary. +- **Monitor the role, not just health:** each interval it reads the primary's `currentRole`. + A node that stops responding **or** is silently demoted to `REPLICA` out-of-band both count + as a failure, so it reacts to a demotion even while the node is still reachable. +- **Grace period:** after detecting loss it waits `QDB_WD_GRACE_PERIOD` seconds (default `5`, + `0` disables) before acting, then re-checks. This lets an operator do a manual switch (demote + the primary, promote a replica) without the watchdog racing them, as long as it completes + within the window. +- **Adopt if already switched:** it then re-checks the cluster; if another node is already + `PRIMARY` (an operator switch, or the old primary recovered) it simply adopts it, no new switch. +- **Step down first:** otherwise it sends a best-effort `{"role":"replica"}` to the old primary, + so a false positive (it is actually still up) cannot leave two primaries. +- **Promote the first healthy node in order:** it then promotes the first **healthy** node in + list order (preferred/earliest first, skipping the dead one), retrying with exponential + backoff. Unreachable candidates are skipped. With `a,b,c`, if `b` is primary and dies it fails + back to `a` when `a` is healthy, and only reaches `c` if `a` is unhealthy or cannot be promoted. + It then monitors the new primary. +- **Never exits on its own:** if no node can be promoted (whole cluster down) it keeps retrying + every interval until one can take over. It stops only on `SIGINT`/`SIGTERM`. All endpoints are `https` (Enterprise only). @@ -276,8 +415,33 @@ export QDB_REST_TOKEN=... QDB_WD_SERVERS=... QDB_WD_FAIL_THRESHOLD=2 ``` The script auto-loads a config file if present (first of `/etc/qdb-primary-watchdog.env`, -`~/.config/qdb-primary-watchdog.env`, or one beside the script; or `$QDB_WD_CONFIG`). See -[`qdb-primary-watchdog.env.example`](./qdb-primary-watchdog.env.example) for every setting. +`~/.config/qdb-primary-watchdog.env`, or one beside the script; or `$QDB_WD_CONFIG`). +[`qdb-primary-watchdog.env.example`](./qdb-primary-watchdog.env.example) is a copy-ready +template of the same settings. + +**Required (no defaults):** + +| Variable | Purpose | +| --- | --- | +| `QDB_REST_TOKEN` | Bearer token for the lifecycle REST API. | +| `QDB_WD_SERVERS` | Ordered `host:9003,...` node list. The first CLI argument overrides it. | + +**Optional (defaults shown):** + +| Variable | Default | Purpose | +| --- | --- | --- | +| `QDB_WD_CONFIG` | *(auto-discover)* | Explicit path to the config file to load, bypassing discovery. | +| `QDB_WD_TARGET_ROLE` | `primary` | Role a node is switched to on failover. | +| `QDB_WD_SWITCH_TIMEOUT_MS` | `5000` | `timeout_ms` in the switch payload, and the promote backoff base. | +| `QDB_WD_STEPDOWN_CONNECT_TIMEOUT` | `3` | Connect timeout (s) for the old-primary step-down; no overall timeout. | +| `QDB_WD_CHECK_CONNECT_TIMEOUT` | `2` | Connect timeout (s) for each role/health poll. | +| `QDB_WD_CHECK_MAX_TIME` | `3` | Overall max time (s) for each role/health poll. | +| `QDB_WD_FAIL_THRESHOLD` | `3` | Consecutive non-primary reads before failover. | +| `QDB_WD_CHECK_INTERVAL` | `1` | Seconds between role checks. | +| `QDB_WD_GRACE_PERIOD` | `5` | Seconds to wait after detecting loss before acting (`0` disables), so a manual operator switch can settle; re-checked after. | +| `QDB_WD_SWITCH_RETRIES` | `3` | Promote attempts, with exponential backoff between them. | +| `QDB_WD_SWITCH_POLL_INTERVAL` | `1` | Seconds between lifecycle polls after a switch. | +| `QDB_WD_SWITCH_POLL_MAX` | `30` | Max polls to wait for a switch to settle. | ### Run it as a service @@ -296,7 +460,8 @@ sudo systemctl enable --now qdb-primary-watchdog journalctl -u qdb-primary-watchdog -f ``` -`Restart=always` brings it back on crash or after an all-nodes-down exit. +`Restart=always` brings it back on crash (the watchdog no longer exits on its own, even when the +whole cluster is down; it keeps retrying). **Run exactly one watchdog per cluster.** Two instances would try to promote different nodes and fight (split-brain). Put it on a separate monitoring host or one designated node. diff --git a/pom.xml b/pom.xml index 540eb70..5024d7a 100644 --- a/pom.xml +++ b/pom.xml @@ -10,13 +10,15 @@ 11 11 - - 1.3.2 + + 1.3.6-SNAPSHOT diff --git a/qdb-primary-watchdog.env.example b/qdb-primary-watchdog.env.example index fdca2cf..0a43f1b 100644 --- a/qdb-primary-watchdog.env.example +++ b/qdb-primary-watchdog.env.example @@ -19,9 +19,12 @@ QDB_WD_SERVERS=172.31.42.41:9003,172.31.41.35:9003,10.0.0.8:9003 #QDB_WD_TARGET_ROLE=primary # role a node is switched to on failover #QDB_WD_SWITCH_TIMEOUT_MS=5000 # switch payload timeout_ms, and the promote backoff base #QDB_WD_STEPDOWN_CONNECT_TIMEOUT=3 # connect timeout (s) for the old-primary step-down; no overall timeout -#QDB_WD_FAIL_THRESHOLD=3 # consecutive no-responses before failover -#QDB_WD_CHECK_INTERVAL=1 # seconds between health checks -#QDB_WD_STARTUP_RETRIES=3 # sweeps to find the initial primary before giving up +#QDB_WD_CHECK_CONNECT_TIMEOUT=2 # connect timeout (s) for each role/health poll +#QDB_WD_CHECK_MAX_TIME=3 # overall max time (s) for each role/health poll +#QDB_WD_FAIL_THRESHOLD=3 # consecutive non-primary reads before failover +#QDB_WD_CHECK_INTERVAL=1 # seconds between role checks +#QDB_WD_GRACE_PERIOD=5 # wait (s) after detecting loss before acting, so a manual + # operator switch can settle; re-checked after (0 disables) #QDB_WD_SWITCH_RETRIES=3 # promote attempts (exponential backoff between them) #QDB_WD_SWITCH_POLL_INTERVAL=1 # seconds between lifecycle polls after a switch #QDB_WD_SWITCH_POLL_MAX=30 # max polls to wait for a switch to settle diff --git a/qdb-primary-watchdog.sh b/qdb-primary-watchdog.sh index f8dde65..b2674ce 100755 --- a/qdb-primary-watchdog.sh +++ b/qdb-primary-watchdog.sh @@ -4,11 +4,17 @@ # Watches an Enterprise cluster for primary failure and promotes the next node. # # Give it the ordered list of lifecycle admin endpoints (host:9003) of ALL nodes. On start it -# finds which one currently reports role PRIMARY and health-monitors it. When that primary -# stops responding, it promotes the first reachable node in list order (earliest first, -# skipping the failed one) via POST /lifecycle/switch, waits for the switch to settle, then -# monitors the newly promoted node. If no node can be promoted (the whole cluster is down), -# it exits. +# finds which one currently reports role PRIMARY and monitors it; if none is primary yet it +# promotes the first healthy node in list order, so a freshly-started cluster gets a primary. +# (Split-brain is not a concern: QuestDB stops the offending node if two ever claim primary.) +# Monitoring is role-based, not +# just a health ping: each interval it reads the node's currentRole, so it reacts both to the +# primary going unreachable AND to it being silently demoted to a replica out-of-band. When the +# primary is lost it first re-checks whether the cluster already has a primary (e.g. an operator +# switch) and adopts it; otherwise it asks the old primary to step down and promotes the first +# HEALTHY node in list order (preferred first, skipping the failed one) via POST /lifecycle/switch, +# waits for the switch to settle, then monitors the new primary. It never exits on its own: if no +# primary exists yet, or none can be promoted, it keeps retrying every check interval. # # Required, no defaults: the node list and QDB_REST_TOKEN. The node list comes from the first # CLI arg or $QDB_WD_SERVERS; the token from $QDB_REST_TOKEN. Everything else has a default @@ -70,9 +76,13 @@ SERVERS_CSV="${1:-${QDB_WD_SERVERS:-}}" TARGET_ROLE="${QDB_WD_TARGET_ROLE:-primary}" # role to switch a node to on failover SWITCH_TIMEOUT_MS="${QDB_WD_SWITCH_TIMEOUT_MS:-5000}" # timeout_ms in the switch payload STEPDOWN_CONNECT_TIMEOUT="${QDB_WD_STEPDOWN_CONNECT_TIMEOUT:-3}" # connect timeout (s) for the old-primary step-down -FAIL_THRESHOLD="${QDB_WD_FAIL_THRESHOLD:-3}" # consecutive health failures before failover -CHECK_INTERVAL="${QDB_WD_CHECK_INTERVAL:-1}" # seconds between health checks -STARTUP_RETRIES="${QDB_WD_STARTUP_RETRIES:-3}" # sweeps to find the initial primary before giving up +CHECK_CONNECT_TIMEOUT="${QDB_WD_CHECK_CONNECT_TIMEOUT:-2}" # connect timeout (s) for each role/health poll +CHECK_MAX_TIME="${QDB_WD_CHECK_MAX_TIME:-3}" # overall max time (s) for each role/health poll +FAIL_THRESHOLD="${QDB_WD_FAIL_THRESHOLD:-3}" # consecutive non-primary reads before failover +CHECK_INTERVAL="${QDB_WD_CHECK_INTERVAL:-1}" # seconds between role checks +GRACE_PERIOD="${QDB_WD_GRACE_PERIOD:-5}" # seconds to wait after detecting loss before + # acting, so a manual operator switch can settle + # (re-checked afterwards; 0 disables the wait) SWITCH_RETRIES="${QDB_WD_SWITCH_RETRIES:-3}" # retries if a switch POST is not accepted SWITCH_POLL_INTERVAL="${QDB_WD_SWITCH_POLL_INTERVAL:-1}" # seconds between lifecycle polls after a switch SWITCH_POLL_MAX="${QDB_WD_SWITCH_POLL_MAX:-30}" # max polls to wait for a switch to settle @@ -98,7 +108,7 @@ fi # GET a node's /lifecycle. Prints the raw JSON, or nothing if unreachable. lifecycle_json() { - curl -ksS --connect-timeout 2 --max-time 3 \ + curl -ksS --connect-timeout "$CHECK_CONNECT_TIMEOUT" --max-time "$CHECK_MAX_TIME" \ -H "Authorization: Bearer $QDB_REST_TOKEN" \ "https://$1/lifecycle" 2>/dev/null } @@ -114,17 +124,6 @@ get_role() { fi } -# Health ping. Prints the HTTP code, or 000 if there was no response at all. -get_http_code() { - local code rc - code="$(curl -ksS -o /dev/null -w '%{http_code}' \ - -H "Authorization: Bearer $QDB_REST_TOKEN" \ - --connect-timeout 2 --max-time 3 "https://$1/lifecycle" 2>/dev/null)" - rc=$? - (( rc != 0 )) && code="000" - printf '%s' "$code" -} - # Poll a node's /lifecycle until its switch settles and it reports the target role. wait_switch_complete() { local hostport="$1" want="${TARGET_ROLE^^}" attempt resp nospace @@ -207,14 +206,26 @@ find_primary() { return 1 } -# Promote the first reachable node in list order (earliest wins), skipping the current -# (failed) primary. So with a,b,c where b is primary and dies, it tries a first, then c; -# it only reaches c if a cannot be promoted. Sets FOUND_IDX, returns 0/1. +# Promote the first HEALTHY node in list order (earliest/preferred wins), skipping the current +# (failed) primary. So with a,b,c where b is primary and dies, it tries a first, then c; it only +# reaches c if a is unhealthy or cannot be promoted. Each candidate's role is read first: an +# unreachable node is skipped, a node that is already PRIMARY is adopted as-is (an out-of-band +# switch beat us to it), and a healthy replica is promoted. Sets FOUND_IDX, returns 0/1. failover_from() { - local cur="$1" n="${#SERVERS[@]}" i + local cur="$1" n="${#SERVERS[@]}" i role for (( i=0; i= STARTUP_RETRIES )); then - log "ERROR: no PRIMARY found among the nodes after $STARTUP_RETRIES sweeps. Exiting." - exit 1 - fi - (( attempt++ )) - log "No primary yet, retrying sweep ($attempt/$STARTUP_RETRIES)..." + local cur + # Startup: ensure the cluster has a primary. Adopt an existing one, or promote the first healthy + # node if none is primary yet. If every node is unhealthy, keep retrying -- never exit. + until ensure_primary; do + log "No primary and none could be promoted yet; retrying in ${CHECK_INTERVAL}s (will not exit)..." sleep "$CHECK_INTERVAL" done - cur=$FOUND_IDX + cur=$ENSURED_IDX log "Primary is ${SERVERS[$cur]} (index $cur). Monitoring every ${CHECK_INTERVAL}s (threshold $FAIL_THRESHOLD)." - local fail_count=0 code + local fail_count=0 role while true; do - code="$(get_http_code "${SERVERS[$cur]}")" - # Any HTTP reply means the primary is alive; only a no-response (000) counts as down. - if [[ "$code" == "000" ]]; then - (( fail_count++ )) - log "Primary ${SERVERS[$cur]} unreachable, no HTTP response (fails: $fail_count/$FAIL_THRESHOLD)" - else - (( fail_count > 0 )) && log "Primary ${SERVERS[$cur]} responding again (HTTP $code), resetting" + # Role-based check: the primary counts as healthy only while it still reports PRIMARY. + # An unreachable node (empty role) or a silent demotion to REPLICA both count as failures. + role="$(get_role "${SERVERS[$cur]}")" + if [[ "$role" == "PRIMARY" ]]; then + (( fail_count > 0 )) && log "Primary ${SERVERS[$cur]} healthy again (role=PRIMARY), resetting" fail_count=0 + else + (( fail_count++ )) + log "Primary ${SERVERS[$cur]} not serving as primary (role=${role:-unreachable}) (fails: $fail_count/$FAIL_THRESHOLD)" fi if (( fail_count >= FAIL_THRESHOLD )); then - log "Primary ${SERVERS[$cur]} is down, initiating failover" - demote_to_replica "${SERVERS[$cur]}" # first, try to make the old primary stand down - if failover_from "$cur"; then + log "Primary ${SERVERS[$cur]} lost; initiating failover" + # Grace period: give an operator doing a manual switch (demote the old primary, promote a + # replica) a few seconds to finish before we act, so the watchdog does not race them. After + # the wait we re-check the cluster; if a primary is now present we simply adopt it below. + if (( GRACE_PERIOD > 0 )); then + log " grace period: waiting ${GRACE_PERIOD}s for a manual switch to settle before acting" + sleep "$GRACE_PERIOD" + fi + # The cluster may already have a new primary (operator switch, or cur recovered elsewhere in + # the list): adopt it instead of forcing another switch. + if find_primary; then cur=$FOUND_IDX fail_count=0 - log "New primary is ${SERVERS[$cur]} (index $cur). Monitoring." + log "Cluster already has a PRIMARY: ${SERVERS[$cur]} (index $cur). Monitoring it." else - log "ERROR: no node could be promoted (whole cluster appears down). Exiting." - exit 1 + demote_to_replica "${SERVERS[$cur]}" # best-effort: make the old primary stand down first + if failover_from "$cur"; then + cur=$FOUND_IDX + fail_count=0 + log "New primary is ${SERVERS[$cur]} (index $cur). Monitoring." + else + # No healthy node to promote. Do NOT exit: leave fail_count armed so the next tick + # retries the whole failover, and keep going until some node can take over. + log "No healthy node could be promoted; will keep retrying every ${CHECK_INTERVAL}s (not exiting)" + fi fi fi diff --git a/src/main/java/com/example/sender/CsvParallelSender.java b/src/main/java/com/example/sender/CsvParallelSender.java index cba9d49..05da68d 100644 --- a/src/main/java/com/example/sender/CsvParallelSender.java +++ b/src/main/java/com/example/sender/CsvParallelSender.java @@ -3,6 +3,8 @@ import io.questdb.client.Sender; import io.questdb.client.Sender.LineSenderBuilder; import io.questdb.client.SenderConnectionListener; +import io.questdb.client.cutlass.http.client.WebSocketUpgradeException; +import io.questdb.client.cutlass.qwp.client.QwpAuthFailedException; import io.questdb.client.cutlass.qwp.client.QwpColumnBatch; import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler; import io.questdb.client.cutlass.qwp.client.QwpQueryClient; @@ -44,6 +46,10 @@ public class CsvParallelSender { private static final String DEFAULT_PROTOCOL = "qwp"; private static final String DEFAULT_SENDER_ID = "ha_sender"; private static final String DEFAULT_STORE_FORWARD_DIR = "/tmp/qdb-sf"; + // Upper bound (ms) on a single TCP connect attempt, so a black-holed host fails over fast + // instead of riding the OS connect timeout. 0 disables (falls back to the OS default). + // QWP + probe only; the ILP path is left untouched so --protocol ilp stays identical. + private static final int DEFAULT_CONNECT_TIMEOUT_MS = 3000; // Batch = one auto-flush append (deferred, no commit under transactional mode). // Transaction = batches-per-transaction batches, committed atomically per table // by an explicit flush(). See buildSender()/runWorker(). @@ -52,6 +58,11 @@ public class CsvParallelSender { // Probe (QWP only): poll the latest ingested timestamp on an interval, 0 disables. private static final long DEFAULT_PROBE_INTERVAL_MS = 1000L; private static final String PROBE_QUERY = "select timestamp from trades limit -1"; + // Enterprise lifecycle status of whichever node the query client is currently connected to. + // Returns the LIVE role (columns like current_role / target_role), unlike the QWP handshake + // SERVER_INFO which is only refreshed on a reconnect and so goes stale after an in-place + // primary<->replica switch. The probe runs this each poll to report the true serving role. + private static final String STATUS_QUERY = "switch status"; // Zone for the query client (egress). Biases failover toward same-zone instances on // Enterprise; a no-op on OSS (which advertises no zone). Empty omits the key. private static final String DEFAULT_ZONE = "eu-west-1"; @@ -88,15 +99,34 @@ public static void main(String[] args) throws Exception { // support it and the connection is rejected, so it is off by default. final boolean enterprise = Boolean.parseBoolean(a.getOrDefault("--enterprise", "false")); final String zone = a.getOrDefault("--zone", DEFAULT_ZONE); + // QWP + probe: bound a single TCP connect attempt (0 disables). ILP is left unchanged. + final int connectTimeoutMs = Integer.parseInt(a.getOrDefault("--connect-timeout-ms", + String.valueOf(DEFAULT_CONNECT_TIMEOUT_MS))); - if (!protocol.equals("qwp") && !protocol.equals("ilp")) { - System.err.println("--protocol must be 'qwp' or 'ilp', got: " + protocol); + if (!protocol.equals("qwp") && !protocol.equals("ilp") && !protocol.equals("qwpudp")) { + System.err.println("--protocol must be 'qwp', 'qwpudp', or 'ilp', got: " + protocol); System.exit(2); } + // QWP/UDP is fire-and-forget datagram ingest: the transport rejects authentication + // (the server accepts any connection on the UDP port), so any credentials passed are + // meaningless. Warn rather than fail, so the same command line works across transports. + if (protocol.equals("qwpudp")) { + final boolean hasAnyAuth = (token != null && !token.isEmpty()) + || (username != null && !username.isEmpty()) + || (password != null && !password.isEmpty()); + if (hasAnyAuth) { + System.err.println("[warn] --protocol qwpudp is unauthenticated (UDP accepts any connection);" + + " ignoring --token/--username/--password"); + } + } if (probeIntervalMs < 0) { System.err.println("--probe-interval-ms must be >= 0 (0 disables the probe)"); System.exit(2); } + if (connectTimeoutMs < 0) { + System.err.println("--connect-timeout-ms must be >= 0 (0 uses the OS connect timeout)"); + System.exit(2); + } if (batchSize <= 0) { System.err.println("--batch-size must be > 0"); System.exit(2); @@ -120,14 +150,19 @@ public static void main(String[] args) throws Exception { } final SenderCfg cfg = new SenderCfg(protocol, addrsCsv, token, username, password, retryTimeout, - senderIdBase, storeForwardDir, batchSize, batchesPerTransaction, numSenders, enterprise, zone); + senderIdBase, storeForwardDir, batchSize, batchesPerTransaction, numSenders, enterprise, zone, + connectTimeoutMs); final String conf = buildConf(addrsCsv, token, username, password, retryTimeout); System.out.println("Ingestion started. Protocol: " + protocol + (protocol.equals("qwp") ? " (WebSocket, sender-id=" + senderIdBase + ", store-and-forward=" + storeForwardDir - + ", batch-size=" + batchSize + ", batches-per-transaction=" + batchesPerTransaction + ")" - : "") + + ", batch-size=" + batchSize + ", batches-per-transaction=" + batchesPerTransaction + + ", connect-timeout-ms=" + connectTimeoutMs + ", retry-timeout-ms=" + retryTimeout + ")" + : protocol.equals("qwpudp") + ? " (QWP/UDP datagrams, ingest-only, unauthenticated, no store-and-forward," + + " no failover; query client disabled, batch-size=" + batchSize + ")" + : "") + " | config: " + conf.replaceAll("(token=)([^;]+)", "$1***") .replaceAll("(password=)([^;]+)", "$1***")); @@ -162,8 +197,10 @@ public static void main(String[] args) throws Exception { reporter.setDaemon(true); reporter.start(); - // Probe (QWP only): a separate thread polls the latest ingested timestamp over a - // QWP query client. Same hosts/auth as the senders; it fails over automatically. + // Probe (QWP/WebSocket only): a separate thread polls the latest ingested timestamp + // over a QWP query client. Same hosts/auth as the senders; it fails over automatically. + // Skipped for ilp and for qwpudp: UDP is ingest-only (there is no query path), so the + // equals("qwp") guard deliberately excludes it. final Thread probe = (protocol.equals("qwp") && probeIntervalMs > 0) ? startProbe(cfg, probeIntervalMs) : null; @@ -181,7 +218,7 @@ public static void main(String[] args) throws Exception { try { f.get(); } catch (ExecutionException ee) { - System.err.println("Worker failed: " + ee.getCause()); + System.err.println("Worker failed: " + ee.getCause() + upgradeHint(ee.getCause())); System.exit(1); } } @@ -207,26 +244,35 @@ private static void runWorker( System.out.printf("Sender %d will send %d events%n", senderId, totalEvents); long sent = 0; final boolean isQwp = cfg.protocol.equals("qwp"); - // QWP with a single worker: stamp each row with the current time client-side. A single - // thread's timestamps are monotonic (no O3) and this avoids QWP's per-batch atNow() - // stamping. ILP, or QWP with more than one worker, use atNow() (server-side, O3-safe). - final boolean perRowMicros = isQwp && cfg.numSenders == 1; - // QWP transactional commit cadence: commit every batchSize * batchesPerTransaction - // rows via an explicit flush(). 0 disables periodic commits (ILP path is unchanged). + final boolean isUdp = cfg.protocol.equals("qwpudp"); + // Single worker on a QWP transport (WebSocket or UDP): stamp each row with the current + // time client-side. A single thread's timestamps are monotonic (no O3) and this avoids + // QWP's per-batch atNow() stamping. ILP, or more than one worker, use atNow() + // (server-side, O3-safe). + final boolean perRowMicros = (isQwp || isUdp) && cfg.numSenders == 1; + // Flush cadence, by transport: + // qwp - transactional commit every batchSize * batchesPerTransaction rows (flush commits). + // qwpudp - no transactions; flush every batchSize rows to emit datagrams (bounds the buffer). + // ilp - 0: no explicit mid-loop flush (the HTTP client auto-flushes). final long commitEveryRows = isQwp ? (long) cfg.batchSize * cfg.batchesPerTransaction - : 0L; + : isUdp + ? cfg.batchSize + : 0L; try ( Sender sender = buildSender(cfg, senderId)) { //( Sender sender = Sender.fromConfig(conf)) { final int n = rows.size(); for (long i = 0; i < totalEvents; i++) { TradeRow r = rows.get((int) (i % n)); - // Build row + // Build row. trade_id = -<1-based sequence>, monotonic per sender, so you + // can check completeness/gaps independent of timestamps and (with dedup) get + // idempotent replay. It is a high-cardinality VARCHAR, deliberately NOT a symbol. sender.table("trades") .symbol("symbol", r.symbol) .symbol("side", r.side) .doubleColumn("price", r.price) - .doubleColumn("amount", r.amount); + .doubleColumn("amount", r.amount) + .stringColumn("trade_id", senderId + "-" + (i + 1)); if (timestampFromFile) { Instant ts = Instant.parse(r.timestamp); @@ -263,7 +309,7 @@ private static void runWorker( sender.flush(); System.out.printf("Sender %d finished sending %d events%n", senderId, sent); } catch (Exception e) { - System.err.printf("Sender %d got error: %s%n", senderId, e.toString()); + System.err.printf("Sender %d got error: %s%s%n", senderId, e.toString(), upgradeHint(e)); throw new RuntimeException(e); } } @@ -362,6 +408,24 @@ private static Sender buildSender(SenderCfg cfg, int workerId) { return buildBuilder(cfg.addrsCsv, cfg.token, cfg.username, cfg.password, cfg.retryTimeout).build(); } + if ("qwpudp".equals(cfg.protocol)) { + // ---- QWP/UDP branch ---- + // Fire-and-forget datagrams to the UDP ingest port (:9007 by convention). The + // transport rejects authentication and does not use TLS, store-and-forward, + // failover, or a connection listener (there is no persistent connection). Any + // token/basic auth on the command line was already warned about and is ignored. + String[] udpAddrs = Arrays.stream(cfg.addrsCsv.split(",")) + .map(String::trim) + .filter(s -> !s.isEmpty()) + .toArray(String[]::new); + + LineSenderBuilder u = Sender.builder(Sender.Transport.UDP); + for (String addr : udpAddrs) { + u.address(addr); + } + return u.build(); + } + // ---- QWP (WebSocket) branch ---- String[] addrs = Arrays.stream(cfg.addrsCsv.split(",")) .map(String::trim) @@ -422,15 +486,16 @@ private static Sender buildSender(SenderCfg cfg, int workerId) { case AUTH_FAILED: msg = "auth failed for " + host; break; - case RECONNECT_BUDGET_EXHAUSTED: - msg = "reconnect budget exhausted, giving up"; - break; + // NOTE: no RECONNECT_BUDGET_EXHAUSTED case on purpose. That kind was dropped from + // the client's SenderConnectionEvent.Kind in newer builds (e.g. 1.3.6-SNAPSHOT), so + // switching on it fails to compile there. If a client still emits it, the default + // branch below narrates it generically. case DISCONNECTED: msg = "connection lost to " + host + " (" + cause + "), will retry"; noisy = true; break; case ENDPOINT_ATTEMPT_FAILED: - msg = "endpoint " + host + " failed (" + cause + "), trying next"; + msg = "endpoint " + host + " failed (" + cause + "), trying next" + upgradeHint(event.getCause()); noisy = true; break; case ALL_ENDPOINTS_UNREACHABLE: @@ -457,11 +522,18 @@ private static Sender buildSender(SenderCfg cfg, int workerId) { b.requestDurableAck(true); } + // Bound a single TCP connect so a black-holed host fails over fast (0 = OS default). + if (cfg.connectTimeoutMs > 0) { + b.connectTimeoutMillis(cfg.connectTimeoutMs); + } + return b.storeAndForwardDir(sfPath) .senderId(who) .transactional(true) .connectionListener(connListener) - .reconnectMaxDurationMillis(300_000) + // reconnectMaxDurationMillis is the QWP analog of the ILP retryTimeoutMillis: + // the overall "keep retrying" budget. Driven by --retry-timeout on both transports. + .reconnectMaxDurationMillis(cfg.retryTimeout) .reconnectInitialBackoffMillis(100) .reconnectMaxBackoffMillis(5_000) .autoFlushBytes(524_288) // 512 KiB, under the ~1MB WS frame cap @@ -495,6 +567,10 @@ private static String queryClientConfig(SenderCfg cfg) { if (tls) { sb.append("tls_verify=unsafe_off;"); } + // Bound the probe's TCP connect too, so it fails over as fast as the senders (0 = OS default). + if (cfg.connectTimeoutMs > 0) { + sb.append("connect_timeout=").append(cfg.connectTimeoutMs).append(';'); + } if (addrs.length > 1) { sb.append("failover=on;"); } @@ -546,8 +622,71 @@ public void onFailoverReset(QwpServerInfo info) { orNone(info.getNodeId()), QwpServerInfo.roleName(info.getRole()), orNone(info.getZoneId())); } }; + // `switch status` reports the serving node's LIVE lifecycle role: current_role, plus + // target_role while a switch is in flight. This is authoritative -- unlike the QWP + // handshake role from getServerInfo(), it reflects an in-place promotion/demotion that + // never dropped the read connection. We show current_role as THE role; getServerInfo() + // only supplies node/zone, and its handshake role is a labelled fallback used solely + // when the status query is unavailable (e.g. missing SYSTEM ADMIN). statusDiag says why. + final String[] currentRole = {null}; + final String[] targetRole = {null}; + final String[] statusDiag = {null}; + final long[] lastStatusDiagMs = {0L}; + final QwpColumnBatchHandler statusHandler = new QwpColumnBatchHandler() { + @Override + public void onBatch(QwpColumnBatch batch) { + final int cols = batch.getColumnCount(); + final StringBuilder names = new StringBuilder(); + for (int c = 0; c < cols; c++) { + if (names.length() > 0) { + names.append(','); + } + names.append(batch.getColumnName(c)); + } + batch.forEachRow(row -> { + for (int c = 0; c < cols; c++) { + final String name = batch.getColumnName(c); + if (name == null) { + continue; + } + final String lower = name.toLowerCase(java.util.Locale.ROOT); + if (!lower.contains("role")) { + continue; + } + final String value = row.isNull(c) ? null : row.getString(c); + if (lower.contains("current")) { + currentRole[0] = value; + } else if (lower.contains("target")) { + targetRole[0] = value; + } else if (currentRole[0] == null) { + // A single unqualified "role" column (older servers) is the current one. + currentRole[0] = value; + } + } + }); + if (currentRole[0] == null) { + statusDiag[0] = "'" + STATUS_QUERY + "' returned no current-role column; columns=[" + names + "]"; + } + } + + @Override + public void onEnd(long totalRows) { + } + + @Override + public void onError(byte status, String message) { + statusDiag[0] = "'" + STATUS_QUERY + "' error (status " + status + "): " + message; + } + + @Override + public void onFailoverReset(QwpServerInfo info) { + } + }; while (!Thread.currentThread().isInterrupted()) { latest[0] = Long.MIN_VALUE; + currentRole[0] = null; + targetRole[0] = null; + statusDiag[0] = null; try { client.execute(PROBE_QUERY, handler); if (wasDown[0]) { @@ -557,13 +696,45 @@ public void onFailoverReset(QwpServerInfo info) { if (latest[0] != Long.MIN_VALUE) { // trades designated timestamp is microseconds by default. Instant ts = Instant.EPOCH.plus(latest[0], ChronoUnit.MICROS); + // Ask the serving node for its live role. A status-query failure must not + // look like a connection loss, so swallow it here (currentRole stays null). + try { + client.execute(STATUS_QUERY, statusHandler); + } catch (Exception ex) { + statusDiag[0] = "'" + STATUS_QUERY + "' threw: " + ex; + } + if (currentRole[0] == null && statusDiag[0] == null) { + statusDiag[0] = "'" + STATUS_QUERY + "' produced no row batch and no error" + + " (not a SELECT-style result on the read path?)"; + } + // Role comes from `switch status` (the authoritative live role of the serving + // node). getServerInfo() only supplies node/zone here; its handshake role is a + // labelled fallback used only when the status query is unavailable. final QwpServerInfo si = client.getServerInfo(); - final String served = si != null - ? " served by role=" + QwpServerInfo.roleName(si.getRole()) - + " node=" + orNone(si.getNodeId()) + " zone=" + orNone(si.getZoneId()) - : ""; + final String node = si != null ? orNone(si.getNodeId()) : "(none)"; + final String zone = si != null ? orNone(si.getZoneId()) : "(none)"; + final String served; + if (currentRole[0] != null) { + final boolean switching = targetRole[0] != null + && !targetRole[0].equalsIgnoreCase(currentRole[0]); + served = " served by role=" + currentRole[0] + + (switching ? " (switching -> " + targetRole[0] + ")" : "") + + " node=" + node + " zone=" + zone; + } else { + final String handshake = si != null ? QwpServerInfo.roleName(si.getRole()) : "unknown"; + served = " served by role=" + handshake + " node=" + node + " zone=" + zone + + " (handshake role; live 'switch status' unavailable, may be stale)"; + } System.out.printf("[probe] latest trades timestamp = %s (raw=%d)%s%n", ts, latest[0], served); + // Explain a missing live role at most once per 30s so it does not spam. + if (currentRole[0] == null && statusDiag[0] != null) { + final long now = System.currentTimeMillis(); + if (now - lastStatusDiagMs[0] > 30_000L) { + lastStatusDiagMs[0] = now; + System.out.println("[probe] live role unavailable: " + statusDiag[0]); + } + } } } catch (Exception e) { if (!wasDown[0]) { @@ -577,7 +748,7 @@ public void onFailoverReset(QwpServerInfo info) { } catch (InterruptedException ie) { Thread.currentThread().interrupt(); } catch (Exception e) { - System.err.println("[probe] stopped: " + e.getMessage()); + System.err.println("[probe] stopped: " + e.getMessage() + upgradeHint(e)); } }, "qwp-probe"); t.setDaemon(true); @@ -589,6 +760,36 @@ private static String orNone(String s) { return (s == null || s.isEmpty()) ? "(none)" : s; } + // Turn a failed QWP WebSocket-upgrade throwable into a human-readable hint, using the + // client's TYPED signals rather than string-matching the message: + // - QwpAuthFailedException -> a definitive auth rejection (HTTP 401/403). + // - WebSocketUpgradeException.isRoleMismatch() (HTTP 421) -> the endpoint is not writable + // (a REPLICA / PRIMARY_CATCHUP), i.e. no primary is available among --addrs to accept writes. + // - WebSocketUpgradeException with status 400 -> a missing/malformed auth token (e.g. an + // unset token env var): the server refuses the upgrade before it can return a 401. + // Walks the cause chain (the upgrade failure is usually wrapped). Returns "" otherwise. + private static String upgradeHint(Throwable t) { + for (Throwable c = t; c != null; c = c.getCause()) { + if (c instanceof QwpAuthFailedException) { + final QwpAuthFailedException a = (QwpAuthFailedException) c; + return " -- auth rejected (HTTP " + a.getStatusCode() + "): bad/expired credentials" + + " or missing permission; check --token or --user/--password"; + } + if (c instanceof WebSocketUpgradeException) { + final WebSocketUpgradeException w = (WebSocketUpgradeException) c; + if (w.isRoleMismatch()) { + return " -- endpoint is not writable (role=" + w.getServerRole() + "): no primary" + + " available among --addrs to accept writes"; + } + if (w.getStatusCode() == 400) { + return " -- HTTP 400 on the upgrade: likely a missing or malformed ILP auth token" + + " (e.g. an unset token env var); check --token or --user/--password"; + } + } + } + return ""; + } + private static String buildConf(String addrsCsv, String token, String username, String password, int retryTimeout) { String[] addrs = Arrays.stream(addrsCsv.split(",")) .map(String::trim) @@ -644,6 +845,7 @@ private static Map parseArgs(String[] args) { case "--batch-size": case "--batches-per-transaction": case "--probe-interval-ms": + case "--connect-timeout-ms": case "--enterprise": case "--zone": if (i + 1 >= args.length) { @@ -682,10 +884,12 @@ private static final class SenderCfg { final int numSenders; final boolean enterprise; final String zone; + final int connectTimeoutMs; SenderCfg(String protocol, String addrsCsv, String token, String username, String password, int retryTimeout, String senderIdBase, String storeForwardDir, - int batchSize, int batchesPerTransaction, int numSenders, boolean enterprise, String zone) { + int batchSize, int batchesPerTransaction, int numSenders, boolean enterprise, String zone, + int connectTimeoutMs) { this.protocol = protocol; this.addrsCsv = addrsCsv; this.token = token; @@ -699,6 +903,7 @@ private static final class SenderCfg { this.numSenders = numSenders; this.enterprise = enterprise; this.zone = zone; + this.connectTimeoutMs = connectTimeoutMs; } } } diff --git a/src/main/java/com/example/sender/TelemetryParallelSender.java b/src/main/java/com/example/sender/TelemetryParallelSender.java index 06bdc67..a2a3cd5 100644 --- a/src/main/java/com/example/sender/TelemetryParallelSender.java +++ b/src/main/java/com/example/sender/TelemetryParallelSender.java @@ -522,9 +522,10 @@ private static Sender buildSender(SenderCfg cfg, int workerId) { case AUTH_FAILED: msg = "auth failed for " + host; break; - case RECONNECT_BUDGET_EXHAUSTED: - msg = "reconnect budget exhausted, giving up"; - break; + // NOTE: no RECONNECT_BUDGET_EXHAUSTED case on purpose. That kind was dropped from + // the client's SenderConnectionEvent.Kind in newer builds (e.g. 1.3.6-SNAPSHOT), so + // switching on it fails to compile there. If a client still emits it, the default + // branch below narrates it generically. case DISCONNECTED: msg = "connection lost to " + host + " (" + cause + "), will retry"; noisy = true;