GeezerVison

View reviewed changes

Comment thread

crates/connectors/src/transport/kafka/nonft/test.rs Outdated

		.expect("the sink must report itself healthy again");
		assert_eq!(callbacks.metric("kafka_output_delivery_errors_total"), 0.0);
		assert!(callbacks.metric("kafka_output_queue_full_wait_seconds_total") > 0.0);
		assert!(

mythical-fred-oss Bot Oct 27, 2026

Copy link

Copy Markdown

Contributor

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.

bug, high: This assert fails every time locally (6 out of 6 runs). While the broker is down, ClientContext::error sends BrokerTransportFailure and AllBrokersDown to cb.error with the tag kafka_nonft_producer, so errors is never empty. The fix is to record the tag and assert only that no kafka_nonft_delivery error occurred. With that change the test passed 3 out of 3 runs:

-    errors: Mutex<Vec<String>>,
+    errors: Mutex<Vec<(Option<&'static str>, String)>>,
...
-    fn error(&self, _fatal: bool, error: AnyError, _tag: Option<&'static str>) {
-        self.errors.lock().unwrap().push(error.to_string());
+    fn error(&self, _fatal: bool, error: AnyError, tag: Option<&'static str>) {
+        self.errors.lock().unwrap().push((tag, error.to_string()));
...
-    assert!(
-        callbacks.errors.lock().unwrap().is_empty(),
+    // The down broker raises client-level transport errors; only a delivery
+    // error means a lost message.
+    assert!(
+        !callbacks
+            .errors
+            .lock()
+            .unwrap()
+            .iter()
+            .any(|(tag, _)| *tag == Some("kafka_nonft_delivery")),

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

mythical-fred-oss Bot reviewed

View reviewed changes

Comment thread

crates/connectors/src/transport/kafka/nonft/output.rs Outdated

		Ok(_) => {
		self.metrics
		.delivered_messages
		.fetch_add(1, Ordering::Relaxed);

mythical-fred-oss Bot Oct 27, 2026

Copy link

Copy Markdown

Contributor

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.

pitfall, medium: Every successful delivery clears the unhealthy state, even while the queue is still full. When the broker is slow but still up, a push can wait on a full queue past MAX_POLLING_INTERVAL while deliveries keep arriving. Each queue_full then sets Unhealthy and the next delivery sets Healthy again within milliseconds, so the status flaps every 5 s and the controller rarely shows the stall. The two flags can also race: swap and the callback are not atomic together. If a delivery's report_healthy lands between a report_unhealthy swap and its callback, the controller ends on Unhealthy while unhealthy == false, and the status stays stuck until the next unhealthy→healthy transition. Consider letting only enqueued() clear a queue-full stall, so deliveries clear only delivery errors. Alternatively, send the callback while holding a lock.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

mythical-fred-oss Bot reviewed

View reviewed changes

Comment thread

crates/connectors/src/transport/kafka.rs Outdated

		/// controller's output thread, a DBSP aux thread that `RuntimeHandle::kill`
		/// joins, and a queue that a down broker keeps full would otherwise hold the
		/// teardown open for as long as the broker stays down.
		/// Enqueues `record` in `producer`, waiting out a full producer queue.

mythical-fred-oss Bot Oct 27, 2026

Copy link

Copy Markdown

Contributor

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.

tech-debt, low: This adds a second doc header under the existing "Hands record to producer…" paragraph, so rustdoc shows two summaries. Please merge them into one header.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

mythical-fred-oss Bot commented Oct 27, 2026

Copy link

Copy Markdown

Contributor

What I ran

Command	Result
`cargo test -p feldera-types output_message_timeout`	✅ pass
`cargo test -p dbsp_connectors --lib send_tests`	✅ pass
`cargo test -p dbsp_connectors --lib nonft::test`	❌ failed 6 out of 6 runs at `test.rs:165` (see inline). Passed 3 out of 3 runs with the suggested fix
`pre-commit run --files <changed>`	✅ clippy, fmt, machete, docs. The OpenAPI hook failed only because of a network download in my sandbox

Gates

Tests, high: the mock-cluster integration test fails as written, so CI will be red. Please fix it before merge.
Docs, medium: this changes a user-visible default (message.timeout.ms becomes 0, so the pipeline now stalls instead of dropping messages) and adds 9 metrics, but only the changelog is updated. The Kafka sink page (docs/connectors/sinks/kafka.md) is deferred to a follow-up PR. Since the default flips in this PR, the sink page should document the new default and the way to restore a bound in this PR too.

Tests not covered by this PR

Delivery failure path: a bounded message.timeout.ms with the broker down should raise kafka_output_delivery_errors_total, report the sink unhealthy with the delivery description, and report it healthy again after the next successful delivery. Nothing exercises report_unhealthy from delivery() today.
Shutdown while stalled: cancel the token while the queue is full. The push should return an error, and the drop-time purge then bumps delivery_errors and reports the sink unhealthy during teardown. Is that intended?
A slow broker, not a down one, to show the health flapping described inline.

Open question (default): I'm in favour of the unbounded default, given that kafka_send already ends on shutdown and the FT sink is unaffected because it never calls validate().

All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

bestouff force-pushed the issue-7578-step2 branch from 5cc4833 to e47572a Compare

October 27, 2026 17:25

bestouff changed the title ~~[connectors] Kafka output: retry delivery without bound, show the queue~~ [connectors] Retain Kafka output through broker outages

[connectors] Give transport output endpoints a health and metrics channel #7639

This was referenced Oct 27, 2026

Merged

[connectors] Replay the transaction in the exactly-once Kafka output #7682

Merged

[docs] Link the Kafka output delivery guarantees from the pages that need them #7683

Merged

bestouff added this pull request to stack #7686

October 28, 2026 19:05

bestouff requested a review from blp

October 28, 2026 19:07

bestouff marked this pull request as ready for review

October 28, 2026 19:17

blp commented Oct 28, 2026

Copy link

Copy Markdown

Member

@bestouff If you like dividing changes into stacks, that's fine, but it would also be fine to submit PRs with more commits in them.

All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

bestouff commented Oct 28, 2026

Copy link

Copy Markdown

Contributor Author

Honestly it was also an excuse to try stacked PRs ...

blp reacted with thumbs up emoji

All reactions

👍 1 reaction

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

bestouff force-pushed the issue-7578-step2 branch from d68c243 to 73a3ecd Compare

October 28, 2026 19:31

mythical-fred-oss Bot commented Oct 28, 2026

Copy link

Copy Markdown

Contributor

fred-review did not finish; see https://github.com/feldera/feldera/actions/runs/40752930124

All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

mythical-fred-oss Bot reviewed

Oct 28, 2026

View reviewed changes

Comment thread

crates/connectors/src/transport/kafka/nonft/output.rs Outdated

		.client()
		.fetch_metadata(Some("test_topic"), Duration::from_secs(10))
		.unwrap();
		endpoint.push_buffer(b"recovered").unwrap();

mythical-fred-oss Bot Oct 28, 2026

Copy link

Copy Markdown

Contributor

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.

bug / high — flaky test. It failed 5 times in 13 runs locally (output.rs:708, "a successful delivery must restore health"). I added a diagnostic and every failure showed delivered=0 errors=3: the recovered record also misses the 1 s message.timeout.ms, because the producer can still be reconnecting after fetch_metadata returns. Health then stays unhealthy for good. Retrying the push until one record is delivered passed 14 times in 14 runs:

Suggested change

	endpoint.push_buffer(b"recovered").unwrap();
	// The reconnect can still outlast the 1 s deadline; retry until one
	// record is delivered.
	wait(
	\|\| {
	let errors = callbacks.metric("kafka_output_delivery_errors_total");
	endpoint.push_buffer(b"recovered").unwrap();
	wait(
	\|\| {
	callbacks.metric("kafka_output_delivered_messages_total") > 0.0
	\|\| callbacks.metric("kafka_output_delivery_errors_total") > errors
	},
	15_000,
	)
	.unwrap();
	callbacks.metric("kafka_output_delivered_messages_total") > 0.0
	},
	90_000,
	)
	.expect("a record must be delivered after the broker is back");
	assert!(matches!(
	callbacks.health_status(),
	Some(ConnectorHealthStatus::Healthy)
	));

(Replace the push_buffer and the wait(...) that follows it. A 5 s inner wait was too short: an expiry sometimes took longer than 5 s to show up.)

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

mythical-fred-oss Bot reviewed

Oct 28, 2026

View reviewed changes

Comment thread

crates/connectors/src/transport/kafka/nonft/output.rs

		// This value was established empirically.
		const MAX_MESSAGE_OVERHEAD: usize = 64;

		const DELIVERY_STALL_TIMEOUT: Duration = Duration::from_secs(10);

mythical-fred-oss Bot Oct 28, 2026

Copy link

Copy Markdown

Contributor

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.

pitfall / medium. The 10 s threshold is hard-coded and measures age in the queue, not lack of progress. A sink limited by broker throughput, with a large queue (the default queue.buffering.max.kbytes is about 1 GiB), can routinely hold messages for more than 10 s. It would then report unhealthy, or flap, while it is delivering normally. Options: make the threshold configurable, or count only "no delivery completed in N s while messages are pending" as a stall.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

mythical-fred-oss Bot commented Oct 28, 2026

Copy link

Copy Markdown

Contributor

Review: unbounded retries and health/metrics for the non-FT Kafka sink

What I ran (debug build, merge commit):

Command	Result
`cargo test -p feldera-types timeout_tests`	3/3 pass
`cargo test -p dbsp_connectors --lib -- transport::kafka::nonft transport::kafka::send_tests`	8/8 on the first run
`enqueue_after_delivery_timeout_does_not_restore_health`, repeated	failed 5 of 13 runs; with the fix in the inline suggestion, 14/14 pass

Flaky test — high, gate failed (zero tolerance for flaky tests). Must be fixed before merge. Root cause and fix are in the inline comment.
Other gates pass. Breaking change declared, with a migration path. Docs and changelog updated. No new dependencies, no unsafe. The new message.timeout.ms=0 default reaches only the non-FT path: the FT endpoint never calls validate().
Missing tests — medium:
1. No test for a non-retriable enqueue error, e.g. MessageSizeTooLarge: the endpoint should become unhealthy and no pending entry should leak.
2. No test checks that the AllBrokersDown and BrokerTransportFailure callbacks no longer reach cb.error. The outage test checks this only indirectly, through errors.is_empty().
Pitfall — medium: on shutdown during an outage, rdkafka's BaseProducer::drop purges the queue. Each purged record goes through complete_delivery, which calls cb.error and increments delivery_errors. With unbounded retries, a full queue (up to 100k records) can produce that many error callbacks at teardown. Consider suppressing Purged* errors once shutdown is cancelled.

All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

bestouff force-pushed the issue-7578-step2 branch from 73a3ecd to 82c3f90 Compare

October 28, 2026 21:20

blp approved these changes

Oct 28, 2026

View reviewed changes

blp left a comment

Copy link

Copy Markdown

Member

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.

I read all the Rust code changes, except the changes to the tests, and it looks good to me.

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

bestouff reacted with thumbs up emoji

All reactions

👍 1 reaction

Comment thread

crates/connectors/src/transport/kafka/nonft/output.rs

Comment on lines +161 to +170

		/// Pending deliveries include the record currently waiting for queue space.
		/// IDs preserve enqueue order even when acknowledgements arrive out of order.
		#[derive(Default)]
		struct DeliveryState {
		next_id: usize,
		pending: BTreeMap<usize, Instant>,
		failure: Option<String>,
		fatal_error: Option<String>,
		unhealthy: bool,
		}

blp Oct 28, 2026

Copy link

Copy Markdown

Member

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.

Needs better comments (probably should have per-field comments).

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

All reactions

bestouff force-pushed the issue-7578-step2 branch from 82c3f90 to e270e53 Compare

October 29, 2026 13:00

bestouff added this pull request to the merge queue

github-merge-queue Bot removed this pull request from the merge queue due to a conflict with the base branch

Base automatically changed from issue-7578 to main

October 29, 2026 15:40


          [connectors] Log Kafka broker disconnections as warnings

6d95c35

The non-fault-tolerant Kafka sink stopped reporting broker transport
failures as connector errors, because librdkafka reconnects without
losing queued records and the health status reports a sustained outage.
It logged them at debug level, so the log at its default level did not
say why a pipeline parked behind the sink. Log them as warnings under the
endpoint's span, and say so in the connector docs and the changelog.

Signed-off-by: Xavier Bestel <[email protected]>

bestouff force-pushed the issue-7578-step2 branch from e270e53 to 6d95c35 Compare

October 29, 2026 15:40

bestouff added this pull request to the merge queue

bestouff removed this pull request from the merge queue due to a manual request

mihaibudiu added this pull request to the merge queue

github-merge-queue Bot removed this pull request from the merge queue due to failed status checks

Uh oh!

There was an error while loading. Please reload this page.

mythical-fred-oss Bot mentioned this pull request

Flaky: a_panic_on_the_worker_task_still_shuts_the_runtime_down_in_time hangs past its shutdown-timeout margin #7716

Closed

mythical-fred-oss Bot commented Oct 30, 2026

Copy link

Copy Markdown

Contributor

Verdict: flaky test, not this PR. High confidence.

The merge queue run failed only one job, Adapter Tests, on:

integrated::postgres::cdc_input::tests::a_panic_on_the_worker_task_still_shuts_the_runtime_down_in_time

job: https://github.com/feldera/feldera/actions/runs/40986438451/job/123845019033 (log lines 1614-1696)

This test deliberately panics a worker task and then expects block_on_then_shut_down to shut the runtime down within RUNTIME_SHUTDOWN_TIMEOUT + 5s; it failed because the wrapper thread didn't finish in time (the wrapper thread hung on the runtime after a panic: Timeout), not because of anything in this PR's diff. This PR only touches crates/connectors/src/transport/kafka/nonft/output.rs and a docs page — no overlap with cdc_input.rs or the postgres CDC path.

Already tracked as a known flake in #7716, which has the same failure from an earlier, unrelated run. I added this run as a new occurrence there.

Next step: re-queue — this should pass on a retry.

All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

mihaibudiu added this pull request to the merge queue

Merged via the queue into main with commit 69bcd41

1 check passed

Uh oh!

There was an error while loading. Please reload this page.

mihaibudiu deleted the issue-7578-step2 branch

October 30, 2026 06:20

yermakoffivan pushed a commit to yermakoffivan/feldera that referenced this pull request