GeezerVison

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",

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"] }

tokio-websockets = { version = "0.13.3", default-features = false, features = ["aws_lc_rs"] }

base64 = "0.22.1"

base58 = "0.2.0"

binrw = "0.15.2"

"rustls-0_23-webpki-roots",

] }

async-nats = { workspace = true }

tokio-websockets = { workspace = true }

async-stream = { workspace = true }

data-encoding = { workspace = true }

anyhow = { workspace = true, features = ["backtrace"] }

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]

// 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!(

fn consumer_name_prefix(consumer_config: &cfg::ConsumerConfig, table_name: &str) -> String {

let base = consumer_config

.name

}

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)

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;

}

}

const DELAYED_CREATE_RECHECK_DELAY: Duration = Duration::from_secs(2);

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;

});

// 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);

}

}

.unwrap();

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();

/// 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>,

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 {

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

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).

- Delta Lake input connector: `uc://` tables can use Databricks workload

| `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`. |

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`:

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