GeezerVison

The World Wide Web, restored to 800 pixels of glory.

GeezerVison is showing github.com — [connectors] Make pause take effect more promptly. by blp · Pull Request #7941 · feldera/feldera

Conversation

blp commented Nov 11, 2026 •

edited

Loading

Uh oh!

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

Copy link

Copy Markdown

Member

Fixes #7568.

A paused single-host pipeline continued to poll its input connectors. Input that connectors buffered before the pause, or shortly after it, still went into the circuit, and no API could confirm that input had stopped. This made tests nondeterministic.

Change

Before	After
A paused pipeline processes all buffered input.	A paused pipeline polls only HTTP input endpoints. These accept input while paused only with `force=true`. Other buffered input waits until resume.
The server reports `Paused` at once, while a step can still take input.	The circuit thread copies the state to `reported_state` between steps. Steps select their input by `reported_state`, and the server reports it. A reported pause thus means that no step takes connector input.

The paused selection is a new StepInputs::Paused variant, so that a coordinator can request it too (second commit). feldera/cloud#2208 uses it so that a paused multihost pipeline processes forced HTTP input.

Exceptions:

Until the pipeline first runs (has_run), steps poll all endpoints. No reader has buffered input yet, and the first poll records each connector's resume position, which a checkpoint needs.
To make a checkpoint, a pipeline without fault tolerance still polls the endpoints that block the checkpoint.
With a coordinator, the coordinator decides. The cloud PR for feldera/cloud#2208 makes it step with StepInputs::Paused while paused.

This is an incompatible change. The docs (testing.md, lifecycle.md, changelog) and the docstring of the SDK's pause() describe the new behavior.

Testing

New tests paused_pipeline_does_not_poll_buffered_input, paused_pipeline_polls_until_it_first_runs, paused_pipeline_processes_forced_http_input and multihost_paused_step_polls_only_http_input. Each fails without its change.
/stats reports the same state as /status.
connector_suites for controller::, server::, transport::http and delta_table: all pass, except kafka_server, which needs a Kafka broker.
Not run locally: tests that need Postgres, DynamoDB, Redis or Kafka, and the Python platform tests.

Sorry, something went wrong.

Uh oh!

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

All reactions

blp added bug

Something isn't working

documentation

Improvements or additions to documentation

connectors

Issues related to the adapters/connectors crate

User-facing For PRs that lead to Feldera-user visible changes labels

Nov 11, 2026

blp self-assigned this

Nov 11, 2026

blp requested a review from ryzhyk

November 11, 2026 19:41

blp mentioned this pull request

Nov 11, 2026

[manager] There is no way to check that all input connectors are paused #7568

Open

blp commented Nov 11, 2026

Copy link

Copy Markdown

Member Author

@ryzhyk This is a draft only because I am uncertain whether we want to fix this behavior at all, since it is so long-standing. On the other hand, the behavior is always confusing.

All reactions

Sorry, something went wrong.

Uh oh!

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

mihaibudiu commented Nov 11, 2026

Copy link

Copy Markdown

Contributor

Sounds like a useful feature

All reactions

Sorry, something went wrong.

Uh oh!

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

mythical-fred-oss Bot reviewed

Nov 11, 2026

View reviewed changes

Comment thread

crates/connectors/src/server.rs Outdated

		let state = match global_metrics.get_state() {
		PipelineState::Terminated => PipelineState::Terminated,
		_ => global_metrics.reported_state(),
		};

mythical-fred-oss Bot Nov 11, 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 (medium): /status now reports reported_state, but /stats (ExternalGlobalControllerMetrics.state, controller/stats.rs ~L1706) still serializes the requested state. A client that polls pipeline.stats().global_metrics.state == PAUSED sees Paused while a step can still take connector input, so it does not get the guarantee this PR adds, and the two endpoints disagree for up to one step. Suggest serializing reported_state() there too, except for Terminated as here.

Sorry, something went wrong.

Uh oh!

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

All reactions

blp Nov 11, 2026

Copy link

Copy Markdown

Member Author

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.

Fixed. GlobalControllerMetrics::reported_state() now returns Terminated once the pipeline terminates, and /stats serializes it, as /status does. The two endpoints now use the same rule.

-- Claude Code, replying at @blp's request.

Sorry, something went wrong.

Uh oh!

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

All reactions

mythical-fred-oss Bot reviewed

Nov 11, 2026

View reviewed changes

Comment thread

python/feldera/pipeline.py

		A paused pipeline does not process the input that its connectors
		buffered, except forced input to HTTP input connectors. When this
		function waits, it returns after the step that was in progress at the
		time of the pause is complete.

mythical-fred-oss Bot Nov 11, 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): wait_for_idle() returns only when total_input_records == total_processed_records. A paused pipeline now holds buffered input, and the PR says connectors can still buffer for a short time after the pause, so pause(); wait_for_idle() (the pattern in tests/workloads/test_now.py) can now hang until its timeout. Please say this here and in the wait_for_idle docstring and testing.md, or have wait_for_idle compare total_processed + buffered while paused.

Sorry, something went wrong.

Uh oh!

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

All reactions

blp Nov 11, 2026

Copy link

Copy Markdown

Member Author

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.

Documented in the wait_for_idle docstring and in testing.md. I did not change wait_for_idle to count buffered input while paused: then it would also return early while the pipeline runs with input that is not yet processed, unless it first checked the status, and "idle" would mean different things in the two states.

tests/workloads/test_now.py is not affected: it waits for buffered_input_records == 0 before it pauses, and its datagen input is finite.

-- Claude Code, replying at @blp's request.

Sorry, something went wrong.

Uh oh!

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

All reactions

blp Nov 11, 2026

Copy link

Copy Markdown

Member Author

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.

Update: on reflection, I changed wait_for_idle after all. It now reads state and buffered_input_records from the same /stats response. For a paused pipeline, it counts buffered input as processed (input == processed + buffered), and it resets the idle interval when the buffered count changes. A running pipeline uses the old rule. /stats now reports the same state as /status, so a reported pause means that no step takes that input. New unit tests: python/tests/unit/test_wait_for_idle.py. The paused case fails with the old rule.

-- Claude Code, replying at @blp's request.

Sorry, something went wrong.

Uh oh!

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

All reactions

mythical-fred-oss Bot commented Nov 11, 2026

Copy link

Copy Markdown

Contributor

What I ran (debug, static-connectors --test connector_suites)

Suite	Result
`controller::`	121/121 pass
`server::`	38/39 pass. `kafka_server` fails because it needs a broker
`paused_pipeline_does_not_poll_buffered_input` ×30	30/30 pass, no flakes

missing (medium): forced HTTP input while paused. Only kafka_server, which needs a broker, tests that a paused single-host pipeline still processes force=true HTTP input. If the is_http_input() exemption broke, normal CI would not catch it. I wrote this test and it passes (3/3 runs):

+    /// A paused pipeline still processes forced HTTP input, and `/status`
+    /// reports `Paused` throughout.
+    #[engine_test]
+    pub fn paused_pipeline_processes_forced_http_input(engine: &dyn PipelineTestEngine) {
+        actix_web::rt::System::new().block_on(async {
+            crate::ensure_default_crypto_provider();
+            let server = start_test_server(
+                engine,
+                &json!({"name": "test", "inputs": {}, "outputs": {}}),
+                Uuid::new_v4(),
+            )
+            .await;
+            start_pipeline(&server).await;
+            send_input(&server, &test_batches(0, 10)).await;
+            pause_pipeline(&server).await;
+            let CompletionTokenResponse { token } =
+                TestHttpSender::send_stream_deserialize_resp::<CompletionTokenResponse>(
+                    server.post("/ingress/test_input1?force=true"),
+                    &test_batches(10, 10),
+                )
+                .await;
+            wait_for_completion(&server, &token).await;
+            assert_eq!(get_stats(&server).await.global_metrics.total_processed_records, 20);
+            wait_for_status(&server, RuntimeStatus::Paused).await;
+        })
+    }

Other uncovered cases (medium):

A pipeline that has never run (has_run == false) and is paused still polls all endpoints.
A checkpoint or suspend while paused without FT takes the CheckpointBarriers path.

None of these has a test.

Gates: tests, flakiness, docs and the breaking-change changelog entry all pass. The two inline comments cover the /stats vs /status mismatch and the wait_for_idle() hang after a pause.

All reactions

Sorry, something went wrong.

Uh oh!

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

blp force-pushed the inputs-quiescent branch from fb82f18 to 365aa7f Compare

November 11, 2026 20:12

blp added the multihost Related to multihost or distributed pipelines label

Nov 11, 2026

blp force-pushed the inputs-quiescent branch 2 times, most recently from 42413b8 to 2ac9938 Compare

November 11, 2026 22:22

blp commented Nov 11, 2026

Copy link

Copy Markdown

Member Author

Thanks. Changes for the uncovered cases:

Case	Test
Forced HTTP input while paused	`paused_pipeline_processes_forced_http_input` (your test). It fails if paused steps do not poll HTTP endpoints.
Paused before the pipeline first runs (`has_run == false`)	`paused_pipeline_polls_until_it_first_runs`. It fails without the `has_run` check. The Delta Lake suspend/resume tests also failed without it.
Checkpoint while paused, without FT	No new test. `next_step_selection` chooses `CheckpointBarriers` before it looks at the paused state, so this path is the same as before the PR. A test needs a connector that clears its barrier, which the mock connectors cannot do.

The controller::, server:: and transport::http suites pass (161), except kafka_server, which needs a broker.

-- Claude Code, replying at @blp's request.

All reactions

Sorry, something went wrong.

Uh oh!

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

blp added 2 commits

November 11, 2026 15:38


          [connectors] Make pause take effect more promptly.

456f2f2

Before, a paused single-host pipeline continued to poll its input
connectors, so input that they buffered before or shortly after the pause
still went into the circuit.  No API could confirm that input had stopped,
which made tests nondeterministic.

Now, without a coordinator, a paused step polls only HTTP input endpoints
(which accept input while paused only with `force=true`) and, during a
checkpoint, the endpoints that block it.  The circuit thread records which
pause or start it applied, and the pipeline reports `Paused` only after it
applies the pause, so no step in progress can take more input.  A multihost
pipeline already stops at a pause, because its coordinator does not step.

For a paused pipeline, the Python SDK's `wait_for_idle()` counts the input
that connectors hold in buffers as if it were processed, because the
pipeline does not process it until it resumes.

Fixes #7568.

Signed-off-by: Ben Pfaff <[email protected]>


          [types,connectors] Add StepInputs::Paused for a paused pipeline's s…

bcebd9f

…teps.

A paused single-host pipeline polled only HTTP input endpoints by means of
a separate `paused` flag in `StepSelection`, with `inputs` set to `All`.
Such a step recorded a `StepFeed` that left out the endpoints that it did
not poll, so a transaction commit could take them as drained.

Now the selection is `StepInputs::Paused`.  It polls only HTTP input
endpoints and records no `StepFeed`, as for `CheckpointBarriers`.  A
coordinator can request the same selection, so that a paused multihost
pipeline can process forced HTTP input (feldera/cloud#2208).

A pipeline reports the new `paused_step_inputs` capability, because an
older pipeline cannot parse a step request with `StepInputs::Paused`.

Signed-off-by: Ben Pfaff <[email protected]>

blp force-pushed the inputs-quiescent branch from 2ac9938 to bcebd9f Compare

November 11, 2026 23:38

blp commented Nov 11, 2026

Copy link

Copy Markdown

Member Author

Update on the uncovered cases:

Case	Status
Checkpoint while paused, without FT	New test `paused_suspend_polls_barrier_input`. A mock connector holds a barrier, then buffers input past it while paused. The suspend must take that input. The test fails if the paused selection takes precedence over `CheckpointBarriers`.
`pause(); wait_for_idle()`	Fixed in the SDK (see the inline thread).

The full local connector run, with Postgres, Redis, DynamoDB and Redpanda, is still in progress. feldera_adapterlib (73), dbsp_connectors (866) and static_connectors (85) pass.

-- Claude Code, replying at @blp's request.

All reactions

Sorry, something went wrong.

Uh oh!

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

blp commented Nov 12, 2026

Copy link

Copy Markdown

Member Author

End-to-end check on a local k3d cluster. A pipeline has a datagen table (200 rows/s) and an HTTP-only table. The script starts it, pauses, waits 5 s, sends 10 rows with input_json(force=True), calls wait_for_idle(), and resumes.

Hosts	0.375.0 (before)	This build (cloud `e77c31f93`, feldera `bcebd9fe8`)
1	Datagen rows kept arriving while paused (1800 → 2200)	No rows while paused. Forced rows processed in 0.2 s. `wait_for_idle()` returned. Rows resumed after `resume()`.
2	Forced rows not processed within 60 s	Same as 1 host: all checks pass, forced rows in 0.2 s.

-- Claude Code, replying at @blp's request.

All reactions

Sorry, something went wrong.

Uh oh!

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

This branch has not been deployed

No deployments

Reviewers

mythical-fred-oss[bot]

mythical-fred-oss[bot] left review comments

ryzhyk

Awaiting requested review from ryzhyk

At least 1 approving review is required to merge this pull request.

Uh oh!

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

Assignees

blp

Labels

bug

Something isn't working

connectors

Issues related to the adapters/connectors crate

documentation

Improvements or additions to documentation

multihost

Related to multihost or distributed pipelines

User-facing

For PRs that lead to Feldera-user visible changes

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

[manager] There is no way to check that all input connectors are paused

Uh oh!

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

2 participants

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.

You are visitor 01402503 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.

GeezerVison

Conversation

blp commented Nov 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Change

Testing

Uh oh!

blp commented Nov 11, 2026

Uh oh!

mihaibudiu commented Nov 11, 2026

Uh oh!

mythical-fred-oss Bot Nov 11, 2026

Choose a reason for hiding this comment

Uh oh!

blp Nov 11, 2026

Choose a reason for hiding this comment

Uh oh!

mythical-fred-oss Bot Nov 11, 2026

Choose a reason for hiding this comment

Uh oh!

blp Nov 11, 2026

Choose a reason for hiding this comment

Uh oh!

blp Nov 11, 2026

Choose a reason for hiding this comment

Uh oh!

mythical-fred-oss Bot commented Nov 11, 2026

Uh oh!

blp commented Nov 11, 2026

Uh oh!

blp commented Nov 11, 2026

Uh oh!

blp commented Nov 12, 2026

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

blp commented Nov 11, 2026 •

edited

Loading

Uh oh!

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