[Anatolii Yatsuk (tolik0)]

What

For global-cursor streams, ConcurrentPerPartitionCursor.ensure_at_least_one_state_emitted changes in two ways:

• It no longer advances the global cursor when the stream did not complete. Before, a partition router that raised after some partitions were read and closed left every generated partition closed, so the cursor moved to the newest record, and partitions that were never generated started from it on the next sync.

• It no longer emits a state when the stream did not complete and the state holds no checkpoint.

• "Did not complete" means a partition was never closed (its read raised) or slice generation stopped early (the partition router raised).

• "No checkpoint" means the state has no global state, no per-partition states and no parent_state. Such a state is only {"use_global_cursor": ..., "lookback_window": ...}.

• Every other path is unchanged: completed streams still emit (including a stream that read zero records), and failed streams that hold a checkpoint still emit it.

Why

ensure_at_least_one_state_emitted runs when a stream finishes, including when it failed. For a stream that failed before checkpointing anything, it emitted {"use_global_cursor": true, "lookback_window": 0}. The platform persists that state, and the next attempt then starts with a non-empty state that records no progress.

For a StateDelegatingStream, a non-empty state selects incremental_stream. Its cursor finds no value and starts from start_date, so the slices the failed attempt never read are skipped for good, and the retry ends green. The full-refresh walk is never resumed.

This hits airbytehq/airbyte#89581 (source-stripe bank_accounts, a global cursor over a static ListPartitionRouter), and also the connector version it replaces (a SubstreamPartitionRouter without incremental_dependency, whose parent_state is always {}).

Reproduction with the source-stripe manifest from airbytehq/airbyte#89581

Three 30-day slices, each with one customer holding one bank account. Attempt 1 fails on slice 0, then attempt 2 starts from the state attempt 1 emitted.

Failure in attempt 1	State emitted, before	Attempt 2, before	State emitted, after	Attempt 2, after
customers 400	{use_global_cursor: true, lookback_window: 0}	events only; slice 0 never read	none	full walk, all records
customers 5xx, retries exhausted	same	same	none	full walk
child request 5xx, retries exhausted	same	same	none	full walk
none (success)	full state	events	full state	events

Tests

• New unit tests in test_concurrent_perpartitioncursor.py:
  • a partition that is never closed, with no incoming state: no state emitted;
  • a partition router that raises during slice generation: no state emitted;
  • a partition router that raises after two partitions, with a record read in a closed one: no state emitted without incoming state, and the incoming global state is emitted unchanged when there is one;
  • a completed stream with zero records: the state is still emitted;
  • a failed partition with an incoming global state: that state is still emitted.
• The first three fail on main and pass with this change.
• The unit_tests/sources/declarative, unit_tests/connector_builder and unit_tests/sources/concurrent_source suites pass locally.
• The source-stripe bank_accounts tests from fix(source-stripe): read bank_accounts full refresh without per-customer partitions airbyte#89581 pass with this change (21).

Risks

• A failed stream that has no checkpoint now emits no state message. The platform keeps that stream's previous state, which before the fix would have been replaced by an equivalent state with no cursor. A first sync therefore retries from scratch, which is the intent.
• The "at least one state on successful syncs" contract is unchanged, because completed streams always emit.
• Release note: for a global-cursor stream without parent state, the final state was the only commit point. Records read by a failed attempt that saved no progress are no longer followed by a state, so destinations don't commit them and the retry reads them again.
• Per-partition mode (global_substream_cursor: false) is out of scope. Every generated partition's cursor reports its start value, so a failed stream still emits a state. More generally, any partial checkpoint (the first closed partition in per-partition mode, or parent_state in global mode) switches a StateDelegatingStream to its incremental stream; fixing that needs a "full refresh completed" marker in StateDelegatingStream.

🤖 Generated with Claude Code

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@tolik0/skip-empty-state-after-failed-partition#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch tolik0/skip-empty-state-after-failed-partition

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

5 954 tests  +52   5 923 ✅ +52   5m 43s ⏱️ -38s
    1 suites ± 0      31 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 7862272. ± Comparison against base commit 3e85723.

This pull request removes 1 and adds 53 tests. Note that renamed tests count towards both.

unit_tests.test.test_scenario_keys ‑ test_fields_awaiting_a_reader_are_still_unread[bypass_reason]

unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_completed_stream_without_records_when_ensure_state_emitted_then_state_emitted
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_partition_and_incoming_checkpoint_when_ensure_state_emitted_then_state_emitted
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_partition_and_no_checkpoint_when_ensure_state_emitted_then_no_state_emitted
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_slice_generation_after_records_when_ensure_state_emitted_then_global_cursor_not_advanced[incoming_global_state]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_slice_generation_after_records_when_ensure_state_emitted_then_global_cursor_not_advanced[no_incoming_state]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_slice_generation_and_no_checkpoint_when_ensure_state_emitted_then_no_state_emitted
unit_tests.test.test_assertion_messages ‑ test_assert_messages_are_renderable[_acceptance_test_config.py]
unit_tests.test.test_empty_scenarios ‑ test_blank_or_non_string_section_bypass_does_not_count[  ]
unit_tests.test.test_empty_scenarios ‑ test_blank_or_non_string_section_bypass_does_not_count[1]
unit_tests.test.test_empty_scenarios ‑ test_blank_or_non_string_section_bypass_does_not_count[True]
…

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

5 957 tests  +52   5 926 ✅ +52   10m 13s ⏱️ -50s
    1 suites ± 0      31 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 7862272. ± Comparison against base commit 3e85723.

This pull request removes 1 and adds 53 tests. Note that renamed tests count towards both.

unit_tests.test.test_scenario_keys ‑ test_fields_awaiting_a_reader_are_still_unread[bypass_reason]

unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_completed_stream_without_records_when_ensure_state_emitted_then_state_emitted
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_partition_and_incoming_checkpoint_when_ensure_state_emitted_then_state_emitted
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_partition_and_no_checkpoint_when_ensure_state_emitted_then_no_state_emitted
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_slice_generation_after_records_when_ensure_state_emitted_then_global_cursor_not_advanced[incoming_global_state]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_slice_generation_after_records_when_ensure_state_emitted_then_global_cursor_not_advanced[no_incoming_state]
unit_tests.sources.declarative.incremental.test_concurrent_perpartitioncursor ‑ test_given_failed_slice_generation_and_no_checkpoint_when_ensure_state_emitted_then_no_state_emitted
unit_tests.test.test_assertion_messages ‑ test_assert_messages_are_renderable[_acceptance_test_config.py]
unit_tests.test.test_empty_scenarios ‑ test_blank_or_non_string_section_bypass_does_not_count[  ]
unit_tests.test.test_empty_scenarios ‑ test_blank_or_non_string_section_bypass_does_not_count[1]
unit_tests.test.test_empty_scenarios ‑ test_blank_or_non_string_section_bypass_does_not_count[True]
…

♻️ This comment has been updated with latest results.

[ZaneHyattAB]

This looks good. I tested it with the #89581 manifest. After a failed sync, the retry now does the full read instead of skipping.

One question: connections that already saved {"use_global_cursor": true, "lookback_window": 0} from an earlier failure will still skip the full read after this fix. Should we treat that state as empty here, or handle those connections another way?

[Daryna Ishchenko (darynaishchenko)]

The fix is correct for the case it targets: a global-cursor stream that fails before checkpointing no longer emits {use_global_cursor, lookback_window}. I confirmed that end-to-end against main for both failure modes, child partition failure and router/parent failure. The change is non-breaking, and successful syncs (including zero-record ones) still emit state.

Two inline comments cover sibling cases that still produce a non-empty state without full progress.

Rollout note: for global-cursor streams with no parent state, the final state was the only commit point. Records a failed attempt read are now no longer followed by a state, so destinations won't commit them and the retry reads them again. That's the intended trade, but it's worth mentioning in the release notes.

[Anatolii Yatsuk (tolik0)]

ZaneHyattAB thanks for testing it. I'd rather not special-case that state in the factory. A successful sync that reads zero records emits exactly {"use_global_cursor": true, "lookback_window": 0} (test_given_completed_stream_without_records_when_ensure_state_emitted_then_state_emitted), so treating it as empty would rerun the full refresh on every sync for those streams. It also wouldn't rescue anyone: that state only lasts until the next attempt, which reads the incremental stream and saves a real cursor. Connections already hit can't be found from their state; they need a stream refresh.