GeezerVison is showing github.com — [connectors] NATS: use async-nats 0.51.0 by bestouff · Pull Request #7696 · feldera/feldera
[connectors] NATS: use async-nats 0.51.0
#7696
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement . We’ll occasionally send you account related emails.
Already on GitHub?
Sign in
to your account
Uh oh!
There was an error while loading. Please reload this page.
Show file tree
Hide file tree
Changes from all commits
Commits
File filter
Filter by extension
.lock
(1)
.md
(2)
.rs
(4)
.toml
(2)
All 4 file types selected
Only manifest files
Viewed files
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Hide whitespace
Diff view
Hide whitespace
There are no files selected for viewing
18 changes: 10 additions & 8 deletions
18
Cargo.lock
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Original file line number
Diff line number
Diff line change
Expand Up
@@ -67,10 +67,11 @@ async-channel = "2.3.1"
# Its default feature set selects ring; aws-lc-rs is the same choice made
# elsewhere in this workspace. The rest of the list restates the defaults, so
# revisit it when upgrading.
async-nats = { version = "0.47 .0", default-features = false, features = [
async-nats = { version = "0.51 .0", default-features = false, features = [
"server_2_10",
"server_2_11",
"server_2_12",
"server_2_14",
"service",
"aws-lc-rs",
"jetstream",
Expand Down
Expand Up
@@ -104,6 +105,11 @@ aws-lc-rs = "1.18.0"
rsa = { version = "0.9.10", features = ["std"] }
rcgen = { version = "0.14.8", default-features = false, features = ["pem", "aws_lc_rs"] }
tokio-postgres-rustls = { version = "0.14.0", default-features = false, features = ["aws-lc-rs"] }
# Pulled in by async-nats for ws:// server URLs. Its default hashes the
# websocket handshake with sha1_smol; the feature keeps it on aws-lc-rs.
# Keep the version in sync with the one async-nats depends on, or the
# lockfile ends up with two copies and only async-nats's one is used.
tokio-websockets = { version = "0.13.3", default-features = false, features = ["aws_lc_rs"] }
Comment thread
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more .
maybe in comment we can add keep this pinned / in sync with version that async-nates use?
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more .
Done in 96fd8f1
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
base64 = "0.22.1"
base58 = "0.2.0"
binrw = "0.15.2"
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -113,6 +113,8 @@ awc = { workspace = true, features = [
"rustls-0_23-webpki-roots",
] }
async-nats = { workspace = true }
# Only here to select aws-lc-rs for async-nats's websocket handshake.
tokio-websockets = { workspace = true }
async-stream = { workspace = true }
data-encoding = { workspace = true }
anyhow = { workspace = true, features = ["backtrace"] }
Expand Down
Expand Up
@@ -264,7 +266,7 @@ z85 = { workspace = true, optional = true }
etl = { workspace = true, optional = true }
[package.metadata.cargo-machete]
ignored = ["aws-sigv4", "deltalake-catalog-unity"]
ignored = ["aws-sigv4", "deltalake-catalog-unity", "tokio-websockets" ]
[target.'cfg(target_os = "linux")'.dependencies]
# Optional so the SQL test fixture (sql-to-dbsp-compiler/temp), which builds
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -501,7 +501,7 @@ impl NatsReader {
// A replay consumer is single-use: whether the replay
// succeeded or not, the next attempt creates a fresh one.
delete_nats_consumer (&jetstream, &stream_ctx, &consumer_name).await;
delete_consumer_family (&jetstream, &stream_ctx, &consumer_name).await;
result.map_err(|error| {
error.with_context(format!(
Expand Down
Expand Up
@@ -929,9 +929,9 @@ impl NatsReader {
/// Every consumer gets this prefix plus a unique suffix (see
/// [`create_nats_consumer`]), so consumers are always named by us rather than
/// by the server. Server-assigned names are opaque, which makes it impossible
/// to attribute a consumer to a pipeline from the NATS side and, worse, keeps
/// the ordered consumer's self-recreation from ever deleting the consumer it
/// replaces (it deletes by the client-side name, which then never matches) .
/// to attribute a consumer to a pipeline from the NATS side. The client
/// library names its own recreations of a consumer after it, see
/// [`list_consumer_family`] .
fn consumer_name_prefix(consumer_config: &cfg::ConsumerConfig, table_name: &str) -> String {
let base = consumer_config
.name
Expand Down
Expand Up
@@ -1096,6 +1096,71 @@ async fn delete_nats_consumer(
}
}
/// Names the consumers on the stream that descend from `consumer_name`: the
/// consumer itself and the `{consumer_name}_{n}` ones the client library
/// creates when it recreates an ordered consumer after a gap, a lost
/// heartbeat or an external deletion. The library deletes the consumer it
/// replaces, so at most one of them normally exists. The unique suffix of
/// our own names (see [`create_nats_consumer`]) keeps one connector's
/// consumers out of another's family.
async fn list_consumer_family(
jetstream: &jetstream::Context,
stream_name: &str,
consumer_name: &str,
) -> AnyResult<Vec<String>> {
use futures::TryStreamExt;
let stream = jetstream.get_stream_no_info(stream_name).await?;
let recreated_prefix = format!("{consumer_name}_");
let mut names = stream.consumer_names();
let mut family = Vec::new();
while let Some(name) = names.try_next().await? {
if name == consumer_name || name.starts_with(&recreated_prefix) {
family.push(name);
}
}
Ok(family)
}
/// Deletes the consumer and every recreation of it, see
/// [`list_consumer_family`]. Falls back to deleting `consumer_name` alone
/// when the listing fails.
async fn delete_consumer_family(
jetstream: &jetstream::Context,
stream_ctx: &StreamContext,
consumer_name: &str,
) {
let stream_name = stream_ctx.stream_name.as_str();
match tokio::time::timeout(
CONSUMER_DELETE_TIMEOUT,
list_consumer_family(jetstream, stream_name, consumer_name),
)
.await
{
Ok(Ok(family)) => {
if family.is_empty() {
debug!(
"No consumer named '{consumer_name}' or recreated from it is left on stream '{stream_name}'"
);
}
for name in family {
delete_nats_consumer(jetstream, stream_ctx, &name).await;
}
}
Ok(Err(error)) => {
warn!(
"Failed to list the consumers of stream '{stream_name}', deleting only '{consumer_name}': {error:#}"
);
delete_nats_consumer(jetstream, stream_ctx, consumer_name).await;
}
Err(_) => {
warn!(
"Timed out after {CONSUMER_DELETE_TIMEOUT:?} listing the consumers of stream '{stream_name}', deleting only '{consumer_name}'"
);
delete_nats_consumer(jetstream, stream_ctx, consumer_name).await;
}
}
}
/// Grace period before re-checking whether a consumer we just deleted has
/// reappeared.
const DELAYED_CREATE_RECHECK_DELAY: Duration = Duration::from_secs(2);
Expand All
@@ -1108,7 +1173,8 @@ const DELAYED_CREATE_RECHECK_DELAY: Duration = Duration::from_secs(2);
/// apply the create after the delete, leaving an orphan that nobody will
/// ever read and that scans the stream on the server's behalf. A delete
/// sent immediately after would race the same reordering, so this instead
/// waits out a grace period, then deletes the consumer again if it exists.
/// waits out a grace period, then deletes the consumer and any recreation
/// of it (see [`list_consumer_family`]) that exists by then.
///
/// Spawned rather than awaited: the caller (a create path, or a pause
/// tearing down a reader) should not block on this best-effort cleanup.
Expand All
@@ -1122,16 +1188,7 @@ fn clean_up_delayed_create(
let consumer_name = consumer_name.to_string();
tokio::spawn(async move {
tokio::time::sleep(DELAYED_CREATE_RECHECK_DELAY).await;
let reappeared = jetstream
.get_consumer_from_stream::<nats_consumer::Config, _, _>(
&consumer_name,
&stream_ctx.stream_name,
)
.await
.is_ok();
if reappeared {
delete_nats_consumer(&jetstream, &stream_ctx, &consumer_name).await;
}
delete_consumer_family(&jetstream, &stream_ctx, &consumer_name).await;
});
}
Expand Down
Expand Up
@@ -1526,16 +1583,17 @@ impl Canceller {
// The reader task is gone, but the client library may have already
// sent its own recreate of this consumer (on a detected gap or lost
// heartbeat) before cancellation reached it. That recreate can still
// land after our delete below, the same reordering
// `clean_up_delayed_create` exists for, so check for it too.
// land after the cleanup below has looked for it, the same
// reordering `clean_up_delayed_create` exists for, so check for it
// too.
let Canceller {
jetstream,
stream_ctx,
consumer_name,
..
} = self;
async move {
delete_nats_consumer (&jetstream, &stream_ctx, &consumer_name).await;
delete_consumer_family (&jetstream, &stream_ctx, &consumer_name).await;
clean_up_delayed_create(&jetstream, &stream_ctx, &consumer_name);
}
}
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -512,6 +512,42 @@ pub fn deleted_positioned_consumer_does_not_reingest(engine: &dyn ConnectorTestE
.unwrap();
}
/// A consumer the client library recreated must not outlive a pause.
///
/// The library names a recreated ordered consumer `{name}_{n}` and deletes
/// the one it replaces, so the connector must delete the recreated consumer
/// rather than the name it chose itself.
#[engine_test(services(Nats))]
pub fn library_recreated_consumer_is_deleted_on_pause(engine: &dyn ConnectorTestEngine) {
use NatsMockAction::*;
run_nats_mock_test(
engine,
basic_nats_config,
&[
StartNats,
CreateStream,
Publish(2),
CreatePipeline,
Extend,
WaitForRecords(2),
DeleteConsumers,
// The library notices the deletion and recreates the consumer.
WaitForConsumerCount {
count: 1,
timeout: CONSUMER_COUNT_TIMEOUT,
},
Pause,
WaitForConsumerCount {
count: 0,
timeout: CONSUMER_COUNT_TIMEOUT,
},
AssertConsumerCountersMatchServer,
Disconnect,
],
)
.unwrap();
}
/// The consumer counters must agree with the server when a create times out
/// but the server created the consumer anyway.
///
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -36,8 +36,9 @@ pub struct ProxyFaults {
/// Drops the server's reply to the nth consumer create. The server still
/// creates the consumer.
pub drop_create_reply: Option<usize>,
/// Holds the nth consumer create until a delete of the same consumer has
/// reached the server, then forwards the create.
/// Holds the nth consumer create until a delete of the same consumer, or
/// a listing of the stream's consumers, has reached the server, then
/// forwards the create.
pub hold_create_until_delete: Option<usize>,
/// Holds every consumer delete for this long before forwarding it.
pub delay_deletes: Option<Duration>,
Expand Down
Expand Up
@@ -336,6 +337,14 @@ async fn proxy_connection(
state.dropped_replies.insert(reply.to_string());
}
to_server.push(frame.bytes());
// The connector lists the stream's consumers before
// deleting; a held create landing right after that
// listing is the reordering the hold stands for.
if subject.starts_with("$JS.API.CONSUMER.NAMES.")
&& let Some((_, create)) = state.held_create.take()
{
to_server.push(create);
}
}
}
if let Some(gate) = request_gate {
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -125,16 +125,17 @@ enum PublishFailure {
/// Classifies an ack error from `async-nats`.
///
/// `Other` carries server errors the client does not map to a kind (such as
/// an exceeded message size, which the pre-publish check keeps off this
/// path) : retried, because a transient server-side condition also lands
/// here, and the retry loop logs the error text so a permanent one is
/// visible in the logs.
/// `MaxPayloadExceeded` is the client's own size check, which the
/// pre-publish check keeps off this path. `Other` carries server errors the
/// client does not map to a kind : retried, because a transient server-side
/// condition also lands here, and the retry loop logs the error text so a
/// permanent one is visible in the logs.
fn classify(error: jetstream::context::PublishError) -> PublishFailure {
match error.kind() {
PublishErrorKind::StreamNotFound
| PublishErrorKind::WrongLastMessageId
| PublishErrorKind::WrongLastSequence => PublishFailure::Permanent(error.into()),
| PublishErrorKind::WrongLastSequence
| PublishErrorKind::MaxPayloadExceeded => PublishFailure::Permanent(error.into()),
PublishErrorKind::TimedOut
| PublishErrorKind::BrokenPipe
| PublishErrorKind::MaxAckPending
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -26,6 +26,14 @@ Source edition can be found on github.
counts, and queue wait time. See
[Kafka output delivery and metrics](/connectors/sinks/kafka#delivery-retries-and-backpressure).
- NATS input connector: the client library is updated to async-nats 0.51.0,
which fixes spurious missing-heartbeat errors while messages are buffered
and pull consumers looping on `409 Exceeded MaxRequest*`. The `max_batch`,
`max_bytes` and `max_expiry` consumer settings now reach the server, where
earlier releases dropped them. A consumer that the library recreates is
named `<prefix>_<uuid>_<n>`, and the connector deletes it together with its
own. See [consumer lifecycle](/connectors/sources/nats#consumer-lifecycle).
## v0.371.0
- Delta Lake input connector: `uc://` tables can use Databricks workload
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -50,8 +50,8 @@ The connector configuration consists of three main sections:
| `deliver_policy` | variant | Yes | Starting point for reading from the stream (see [Deliver Policy](#deliver-policy)) |
| `max_waiting` | integer | No | Maximum outstanding pull requests. Default: 0 |
| `metadata` | map (string → string) | No | Consumer metadata key-value pairs |
| `max_batch` | integer | No | Maximum messages per batch |
| `max_bytes` | integer | No | Maximum bytes per batch |
| `max_batch` | integer | No | Maximum messages per pull request. The server rejects larger pull requests, and the connector sizes its own accordingly |
| `max_bytes` | integer | No | Maximum bytes per pull request |
| `max_expiry` | duration | No | Maximum time a pull request stays parked on the server, for example `30s`. Replaces the deprecated `max_expires`. |
#### Deliver Policy
Expand Down
Expand Up
@@ -145,14 +145,18 @@ or the table name if none is set. Characters that JetStream does not allow in
consumer names are replaced with `_`. The unique suffix prevents "consumer
already exists" errors on quick restarts, and the prefix identifies which
connector owns a consumer when you inspect the stream with `nats consumer ls`.
For example, to list the consumers of a connector whose prefix is `orders`:
When the client library recreates a consumer after a gap, a lost heartbeat, or
an external deletion, the replacement is named `<prefix>_<uuid>_<n>` and the
replaced consumer is deleted. For example, to list the consumers of a
connector whose prefix is `orders`:
```bash
nats consumer ls <stream> --names | grep '^orders_'
```
The connector deletes its consumer when it pauses, stops, abandons a failed
reader before retrying, or finishes a replay. It also deletes the consumer
The connector deletes its consumer, and any replacement the client library
made for it, when it pauses, stops, abandons a failed reader before retrying,
or finishes a replay. It also deletes the consumer
after a create request that timed out, because an overloaded server may still
create the consumer after the client has given up. Deletion is best-effort: if
the server cannot be reached, the consumer expires on its own after 30 seconds
Expand Down
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.