I like the name of this: "he's so full of bytes!"

indeed, I think counting records is pointless, a record can be arbitrarily large

Using records relies on the tacit assumption that humans know something about the size of the input records for each table.

Worse, it leaves us guessing all the time based on support bundles

Yes. We could reasonably drop max_queued_records except that people actually configure it, so it would break people's pipelines.

A warning would be great for users still using max-queued-records (and nudging them to use the other one instead).

That's a good idea. I can log a warning.

(I don't think we elevate log messages beyond the Log pane, which means that users probably won't see it, though, unless @mihaibudiu wants to eventually make the SQL compiler emit such a warning by parsing the connector configuration, or the web-console starts elevating log warnings/errors.)

I'm adding this:

        if let Some(max_queued_records) = resolved_connector_config.max_queued_records
            && !resolved_connector_config.max_queued_bytes.is_none()
        {
            let max_queued_bytes = resolved_connector_config.max_queued_bytes();
            warn!(
                "Input endpoint {endpoint_name} configures `max_queued_records` to {max_queued_records}, but not `max_queued_bytes`, which will default to {max_queued_bytes}.  We recommend setting `max_queued_bytes` explicitly (`max_queued_records` is less useful and may be omitted)."
            )
        }

Inverted condition. The message says "configures max_queued_records ... but not max_queued_bytes, which will default to ..." — i.e. it is meant to fire when the user set records but left bytes implicit. But the guard is max_queued_bytes.is_some(), so it actually fires only when both are set (and never produces a defaulted value), and stays silent in the case the message was written for. Either flip to max_queued_bytes.is_none() (and then the printed max_queued_bytes is genuinely the implied default), or rewrite the message to match the current condition.

Also: the message ends with (max_queued_records is less useful and may be omitted). That advice contradicts the changelog entry, which tells users to keep max_queued_records and tune max_queued_bytes. Pick one story.

Carried-over typo: "the circuit since the circuit, since the circuit is still busy...". The original max_queued_records doc got this fixed in this same PR (line 1608 now reads cleanly); the new max_queued_bytes doc was copy-pasted from the broken version. Drop the first "since the circuit".

Rustdoc nit: 1000 * max_queued_records is true for the default case, but if the user sets max_queued_records explicitly and leaves max_queued_bytes unset, the implied byte cap is 1000 * <user value>, which can be a very different number from 1 GB and is easy to under-estimate on upgrade. Worth saying explicitly here (the changelog covers it; the field doc should too, since this is what most config readers will look at).

Soft: max_queued_records() as usize (and the bytes counterpart) silently truncate on 32-bit targets. Feldera doesn't run there today, but if someone passes u64::MAX for "unlimited" on a future 32-bit build, this collapses to a small number and re-engages backpressure. A try_into().unwrap_or(usize::MAX) (or just keeping the comparison in u64) is cheap insurance. Same for old.records + increment — if a misconfigured limit lives near usize::MAX, the addition wraps. saturating_add would do.

Naming: is_full_of_records / is_full_of_bytes reads a little odd — is_records_full / is_bytes_full, or records_at_threshold / bytes_at_threshold, is closer to what these actually check (they fire at >=, not when the buffer is literally full). Non-blocking.

max_queued_bytes defaults to 1000 * max_queued_records — fine. The boundary worth pinning explicitly: max_queued_records = u64::MAX (or any value > u64::MAX / 1000) saturates, which max_queued_bytes_default_saturates does cover for max_queued_bytes. Worth adding a one-liner that confirms max_queued_records() itself does NOT saturate so the two getters' semantics are documented side-by-side.

These threading tests rely on recv_timeout(100ms) for the positive case and recv_timeout(50ms) for the negative case. On a loaded CI runner the 50ms negative wait can occasionally be too short to be sure the second batch wouldn't eventually unpark, and 100ms can flake under heavy contention. Consider using parker.park_timeout() directly or, at minimum, bumping the negative wait to ~200ms.

Also: backpressure_unparked_only_on_threshold_crossing doesn't cover the symmetric case for bytes (cross bytes once, then add more bytes — should not re-unpark). Cheap to add.

Worth calling out more sharply that this is a behavior change on upgrade: any pipeline whose average record size is well under 1 kB is unaffected, but Avro/Protobuf/wide-CDC records easily exceed 1 kB and those pipelines will hit the new byte cap earlier than the old record cap and pause sooner. Recommend either (a) explicit "if your records average > 1 kB, expect earlier pauses; mitigation: raise max_queued_bytes", or (b) flip to opt-in (None = no byte limit) for one release.

Minor: "The defaults of 1,000,000 records and 1,000,000,000 bytes limit the memory used by a connector to 1 GB." The two limits are connected (bytes = 1000 * records), so a user reading only this row may not realize that changing max_queued_records silently changes max_queued_bytes too. One sentence — "unless overridden, max_queued_bytes tracks max_queued_records" — would prevent surprises.