fix(bulk-cdk-load): scope state bookkeeping per stream to stop spurious unflushed-state failures

[devin-ai-integration]

What

Destination syncs (reported on destination-snowflake, but the defect is in the bulk-load CDK and affects every bulk destination) intermittently fail at the very end with:

java.lang.IllegalStateException: Sync completed, but unflushed states were detected.

All streams reach COMPLETE and all data is written; only the final checkpoint reconciliation fails, so the sync is marked failed and retried. The diagnostic line in the logs is always the same shape — a single leftover stream state whose expected count was never recorded:

- descriptor=Descriptor(namespace=..., name=...) first key: StateKey(id=1, partitionKeys=[PartitionKey(id=nCJ1)])
  • Incomplete: expectedCount null does not equal flushedCount 0.0

Root cause: StateHistogramStore is a process-wide singleton holding one expected histogram keyed by StateKey(id, partitionKeys) and one flushed histogram keyed by PartitionKey — neither carries stream identity — while the state queues in StateStore are per stream. In socket mode the key material comes straight from the source: id is a per-feed counter (so every stream's first checkpoint is id=1) and partition_id is a 4-character random string. When two streams in one sync draw the same (id, partition_id), their buckets merge; the first stream to reconcile calls remove(key) and deletes the shared expected/flushed entries, after which the second stream's queued state has expectedCount = null forever, isComplete() evaluates null == 0.0 → false, and PipelineRunner fails the sync. The collision is probabilistic, which matches the observed pattern (affected connections succeed on retry and on subsequent syncs).

Resolves https://github.com/airbytehq/oncall/issues/12123:

• https://github.com/airbytehq/oncall/issues/12123

How

Give the bookkeeping the stream identity its keys lack, without changing any key derivation on the source or record side:

• New StateScope — Global or Stream(descriptor).
• StateHistogramStore now holds flushed: Map<Descriptor, PartitionHistogram> and expected: Map<StateScope, StateHistogram>; acceptFlushedCounts, acceptExpectedCounts, isComplete, whyIsStateIncomplete and remove all take the scope (or descriptor). Two streams can no longer touch each other's buckets.
• Global states keep today's exact semantics. A global checkpoint's partition ids are intentionally shared across streams (a global-feed partition id such as outer_partition covers records from several streams), so for StateScope.Global the flushed lookup sums a partition key across all descriptors and remove clears it from all of them.
• The flushed side is keyed by mappedDescriptor (that is AggregateStore's key and what StateStage / PipelineCompletionHandler already have in hand), whereas checkpoints carry the unmapped descriptor, so StateStore injects DestinationCatalog and translates unmapped → mapped. An unknown descriptor falls back to the unmapped one with a warning rather than throwing.

Bulk CDK load version bumped to 1.0.21 with a changelog entry. No connector files change here — the fix reaches destination-snowflake and the other bulk destinations on their next CDK bump.

Breaking change evaluation

Not breaking: no schema, spec, primary key, cursor, stream, or state-format change. The only signature changes are on internal bulk-CDK classes with no external consumers, and global-state accounting is unchanged by construction.

Review guide

1. StateHistogramStore.kt — the scoping itself; check the Global branches preserve cross-stream summing/removal.
2. StateStore.kt — scope derivation per message type and the unmapped → mapped descriptor translation.
3. StateStage.kt / PipelineCompletionHandler.kt — one-line descriptor plumbing.
4. StateHistogramStoreTest.kt / StateStoreTest.kt — new tests.

Test Coverage

New unit tests, all in :airbyte-cdk:bulk:core:bulk-cdk-core-load:

• StateHistogramStoreTest: two Stream scopes sharing an identical StateKey(1, [PartitionKey("abcd")]) stay independent, and removing stream A's key leaves stream B complete. Verified this test fails against the pre-fix unscoped implementation (AssertionError on the post-removal assertion) and passes with the fix.
• StateHistogramStoreTest: Global scope sums flushed counts across descriptors and remove clears them everywhere (regression guard for global mode).
• StateStoreTest: two stream checkpoints for different streams with identical state keys both flush and leave hasStates() == false — i.e. the reported failure no longer occurs.
• StateStoreTest: a stream whose mapped descriptor differs from its unmapped descriptor completes when flushed counts arrive under the mapped descriptor (guards the translation).

The full :airbyte-cdk:bulk:core:bulk-cdk-core-load:test suite passes locally. In CI, the destination-snowflake, s3, s3-data-lake, bigquery, clickhouse, mssql, azure-blob-storage and dev-null integration suites all pass against this branch.

Reproduction of the customer sync itself was not attempted: it needs the customers' source and destination credentials, and the trigger is a random key collision. The unit reproduction above exercises the exact failing invariant instead.

User Impact

Syncs no longer fail spuriously at the end of an otherwise successful run when two streams collide on a checkpoint key. No behavior change for global states, and no change to what is written to the destination. Data was never lost by this bug — only the checkpoint commit and the sync's success status were.

Can this PR be safely reverted and rolled back?

• YES 💚
• NO ❌

Link to Devin session: https://app.devin.ai/sessions/cd8ad0bcacd24e97b2aeea63acf755c1

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

PR Slash Commands

Airbyte Maintainers (that's you!) can execute the following slash commands on your PR:

• 🛠️ Quick Fixes
  • /format-fix - Fixes most formatting issues.
  • /bump-version - Bumps connector versions, scraping changelog description from the PR title.
    • Bump types: patch (default), minor, major, major_rc, rc, promote.
    • The rc type is a smart default: applies minor_rc if stable, or bumps the RC number if already RC.
    • The promote type strips the RC suffix to finalize a release.
    • Example: /bump-version type=rc or /bump-version type=minor
  • /bump-progressive-rollout-version - Alias for /bump-version type=rc. Bumps with an RC suffix and enables progressive rollout.
• ❇️ AI Testing and Review (internal link: AI-SDLC Docs):
  • /ai-prove-fix - Runs prerelease readiness checks, including testing against customer connections.
  • /ai-canary-prerelease - Rolls out prerelease to 5-10 connections for canary testing.
  • /ai-review - AI-powered PR review for connector safety and quality gates.
• 📝 AI Documentation:
  • /ai-docs-review - AI-powered documentation review for PRs with connector changes.
  • /ai-create-docs-pr - Creates a documentation PR for connector changes, stacked on the current PR.
• 🚀 Connector Releases:
  • /publish-connectors-prerelease - Publishes pre-release connector builds (tagged as {version}-preview.{git-sha}) for all modified connectors in the PR.
  • /enable-autopilot-rollouts - Enables autopilot progressive rollouts for the modified connector(s) in the PR, remediating "autopilot rollouts not enabled for {connector-name}" auto-merge blockers. Sets defaultRolloutMode: autopilot and enableProgressiveRollout: true, preserving any existing autopilotConfig.
    • Optional args: connector=<CONNECTOR_NAME> (defaults to the modified connectors in the PR), strategy=fast|slow|default (defaults to fast).
    • Example: /enable-autopilot-rollouts or /enable-autopilot-rollouts connector=source-faker strategy=slow
• ☕️ JVM connectors:
  • /update-connector-cdk-version connector=<CONNECTOR_NAME> - Updates the specified connector to the latest CDK version.
    Example: /update-connector-cdk-version connector=destination-bigquery
• 🐍 Python connectors:
  • /poe connector source-example lock - Run the Poe lock task on the source-example connector, committing the results back to the branch.
  • /poe source example lock - Alias for /poe connector source-example lock.
  • /poe source example use-cdk-branch my/branch - Pin the source-example CDK reference to the branch name specified.
  • /poe source example use-cdk-latest - Update the source-example CDK dependency to the latest available version.
• ⚙️ Admin commands:
  • /force-merge reason="<REASON>" - Force merges the PR using admin privileges, bypassing CI checks. Requires a reason.
    Example: /force-merge reason="CI is flaky, tests pass locally"

📚 Show Repo Guidance

Helpful Resources

• Breaking Changes Guide - Breaking changes, migration guides, and upgrade deadlines
• Developing Connectors Locally
• Managing Connector Secrets
• On-Demand Regression Tests
• #connector-ci-issues
• #connector-publish-updates
• #connector-build-statuses

📝 Edit this welcome message.

[devin-ai-integration]

CI status: 5 failing checks, none of which I believe are caused by this change. Evidence for each:

• Test destination-gcs-data-lake — compileKotlin fails on a deprecated Types.NestedField.of(...) under -Werror. Same failure on bulk-CDK PRs that don't touch state code: #83220, #83201. Iceberg resolves to 1.7.0 on both master and this branch. A dedicated fix is already open in #81445.
• Test destination-postgres — PostgresRawDataDumper.kt:53: No parameter with name 'useFastTimestampParsing', a stale call site left by 4f64b80 (PR fix: mainline fast date time coercion #77539) which removed that parameter from AirbyteValueCoercer. Same failure on #83201; already fixed in #83235.
• Test destination-redshift — every test in the suite, including pure RedshiftSchemaMapperTest unit-style cases, fails with ERROR: exceeded the maximum number (9900) of schemas allowed in database. Test-account exhaustion, not a code failure.
• Test destination-databricks — integration/environment failures (missing Airbyte meta columns, 10-minute timeouts). Also failing on unrelated PRs #82742 and #83201.
• Deploy Docs to Vercel (Preview) — AbortError: The user aborted a request during upload.

Counter-evidence that the CDK change itself is healthy: the destination-snowflake, s3, s3-data-lake, bigquery, clickhouse, mssql, azure-blob-storage and dev-null suites all pass against this branch, and I found no call sites of the changed StateHistogramStore / StateStore APIs outside the files updated here.

---

Devin session

[Airbyte Support Bot (airbyte-support-bot)]

↪️ Triggering /ai-prove-fix per Hands-Free AI Triage Project triage next step.

Reason: Draft PR whose 5 red checks were shown to reproduce on unrelated bulk-CDK PRs; validating the per-stream state bookkeeping fix against live destination connections is the next pipeline gate for:

• https://github.com/airbytehq/oncall/issues/12123

Devin session

[octavia-bot]

🔍 AI Prove Fix session starting... Running readiness checks and testing against customer connections. View playbook

• 🔗 Session URL: https://app.devin.ai/sessions/5baa697c8d4a4858ac4ea2ea43412903

✅ Devin AI session created successfully!

[Airbyte Support Bot (airbyte-support-bot)]

🧪 Fix Validation Evidence

🟢 Outcome: Fix Proven

Isolated testing only — this is a CDK-only PR, so there is no connector image to pre-release and no connection was pinned or synced. I reconstructed the failure independently against pre-fix master (not trusting the PR's claim that its new test fails pre-fix): two checkpoints for two different streams carrying the identical key StateKey(1, [PartitionKey("abcd")]) leave one state permanently unflushable, with the log line expectedCount null does not equal flushedCount 0.0 matching the reported failure signature exactly. On this branch the same scenario flushes both states, and the full bulk-cdk-core-load suite passes (407 tests, 0 failures). The eight destination integration suites that exercise the changed dataflow pipeline end to end are green, and every failing suite is shown to fail identically on unrelated bulk-CDK PRs.

🚦 Next Steps

1. This PR appears ready for review and merge (it is currently a draft).
2. Live-connection validation only becomes meaningful once a bulk destination is bumped onto CDK 1.0.21 — start with destination-snowflake (where the issue was reported) and run /ai-canary-prerelease on that connector PR.
3. Field signal after release: the expectedCount null signature should stop appearing on socket-mode bulk destinations; daily_hands_free_triage will monitor the rollout.

---

📋 Connector & PR Details

Component: bulk-load CDK — airbyte-cdk/bulk/core/load (1.0.20 → 1.0.21); no connector files changed
PR: #83238 (head 4cac3d5)
Pre-release version tested: n/a — CDK-only PR, no publishable connector artifact
Comparison baseline: origin/master (0673d69) source build — regression-test harness (and its override_control_image) is source-connector-only and does not apply here
Detailed results: https://github.com/airbytehq/oncall/issues/12123#issuecomment-5130391522

📝 Evidence Plan

Testing strategy decision

• Regression tests: not applicable. The harness supports sources only; this is a destination-side CDK change with no connector artifact to compare.
• Pre-release + live pinning: not applicable. A CDK-only PR produces no publishable connector image; live validation would need a separate connector PR bumping a destination onto CDK 1.0.21, plus explicit human approval to pin.
• Therefore: isolated testing, with an independently reconstructed before/after pair as the primary proving evidence.

Proving criteria

1. Before (pre-fix master): a locally written test driving StateStore with two checkpoints for two different streams sharing StateKey(1, [PartitionKey("abcd")]) must leave one state unflushable, reproducing expectedCount null does not equal flushedCount 0.0.
2. After (this branch): the same scenario flushes both states; full module suite passes.
3. Global-state semantics unchanged — global scope still sums flushed counts across descriptors and clears them all on removal.
4. No regression in real syncs — destination integration suites green, with every failure shown to reproduce on unrelated bulk-CDK PRs.
5. No stale call sites of the changed StateHistogramStore / StateStore APIs (the StateStore constructor gained a DestinationCatalog parameter).

Disproving criteria

• The before/after pair fails to separate (pre-fix scenario passes ⇒ root cause misdiagnosed; branch scenario still fails ⇒ fix ineffective).
• Global-scope accounting behaviour changes.
• Any destination integration suite regresses relative to unrelated bulk-CDK PRs.
• A caller of the changed APIs is left outside the PR's edits.

Cases attempted

#	Case	Outcome
1	Pre-fix reproduction reconstructed against master's API	✅ fails pre-fix with the exact reported signature
2	Full :airbyte-cdk:bulk:core:bulk-cdk-core-load:test on the branch	✅ 407 tests, 0 failures, 2 skipped
3	Repo-wide call-site sweep for the changed APIs	✅ no stale call sites
4	Destination integration matrix, branch head vs. unrelated bulk-CDK PRs	✅ no regression
5	Live sync on a pinned pre-release	⛔ not applicable (CDK-only PR; probabilistic socket-mode-only trigger)

✅ Pre-flight Checks

• Viability: the diff addresses the diagnosed cause — expected is keyed by StateScope, flushed by mapped stream descriptor, so two streams can no longer share buckets; no key derivation changes on the source/record side.
• Safety: no obfuscation, no new network calls, no credential handling; confined to in-memory bookkeeping, tests, a changelog line and the version file.
• Breaking Change: none — no schema, spec, primary key, cursor, stream, or state-format change. Signature changes are on internal CDK classes, confirmed by the call-site sweep.
• Pin Exclusion: n/a — no connection pinned or synced.
• Reversibility: patch-level CDK bump, in-memory-only state, nothing persisted differently ⇒ clean downgrade; changelog entry present.

Design-intent notes (non-blocking, for reviewers):

1. The unscoped histograms carried no comment suggesting the sharing was intentional, and the PR explicitly preserves the one case where cross-stream sharing is intentional (global checkpoints), covered by a new test. The design intent looks correctly identified.
2. The flushed side is keyed by the mapped descriptor (that is AggregateStore's key). If a mapping ever collapsed two source streams onto one mapped descriptor, those two would still share a bucket — the same collision class, much narrower. Not reachable via normal namespace mapping, but worth a bulk-CDK owner's eye.
3. StateStore.mappedDescriptor() falls back to the unmapped descriptor with a warn when the catalog has no entry. Safe, but if it ever fires it silently reintroduces mixed keying; a maintainer may prefer it to be loud.
4. Global socket-mode accounting in a real sync is the one thing isolated testing cannot fully cover — a bulk-CDK owner sanity-check there is still worthwhile.

📊 Detailed Evidence Log

1. Pre-fix reproduction — scratch worktree at origin/master (0673d69), local-only tests written against the pre-fix API (the PR changes method signatures, so its new tests cannot simply be copied back):

BEFORE_REPRO_INCOMPLETE: expectedCount null does not equal flushedCount 0.0 (by partition: [0.0])
java.lang.AssertionError: Expected value to be false.
1 test completed, 1 failed

hasStates() stays true after the first state reconciles and remove(key) deletes the shared entries — a state that can never flush, which is what PipelineRunner converts into Sync completed, but unflushed states were detected. The equivalent test at the StateHistogramStore level fails the same way.

2. Post-fix — branch head, ./gradlew :airbyte-cdk:bulk:core:bulk-cdk-core-load:test: BUILD SUCCESSFUL, 407 tests, 0 failures, 0 errors, 2 skipped. StateHistogramStoreTest 11/11 (including the colliding-key isolation case and the global-scope regression guard); StateStoreTest 14/14 (including identical stream state keys completing independently and the unmapped → mapped descriptor translation).

3. Destination integration matrix on head 4cac3d5 — pass: snowflake, bigquery, s3, s3-data-lake, clickhouse, mssql, azure-blob-storage, dev-null. Fail: redshift, postgres, gcs-data-lake, databricks, plus the Vercel docs preview — the same four fail with an identical pass/fail split on unrelated bulk-CDK PRs #83201 and #83220, which do not touch state code.

4. Call-site sweep across airbyte-cdk/ and airbyte-integrations/: no unchanged caller of acceptFlushedCounts, acceptExpectedCounts, whyIsStateIncomplete, or the histogram store's isComplete/remove; no destination connector imports, subclasses, or constructs the dataflow StateStore (production instances are DI-provided via InputBeanFactory). The StreamStateStore references in destinations are a different class (io.airbyte.cdk.load.write.StreamStateStore) and are unaffected.

Scope note: the defect is socket-mode only. SelfDescribingStateKeyClient (data-channel.medium=SOCKET) takes both the ordinal and the 4-character partition_id from the source, so two streams can collide; InferredStateKeyClient (STDIO) derives partition keys that embed the stream descriptor and cannot collide.

Connection IDs and customer details are recorded in the linked private issue.

---

Devin session

[Airbyte Support Bot (airbyte-support-bot)]

↪️ Triggering /ai-review per Hands-Free AI Triage Project triage next step.

Reason: /ai-prove-fix returned 🟢 Fix Proven on the current HEAD 4cac3d5f (spurious unflushed-state failure reconstructed on pre-fix master and gone with the per-stream bookkeeping), so the PR has been marked ready for review and the review gate is the next step.

---

Devin session

[octavia-bot]

AI PR Review starting...

Reviewing PR for connector safety and quality.
View playbook

• 🔗 Session URL: https://app.devin.ai/sessions/386ab9388612456c848767d95969f580

✅ Devin AI session created successfully!

[Airbyte Support Bot (airbyte-support-bot)]

🛡️ AI PR Review Report

⚪ Review Action: SKIPPED (OUT OF SCOPE)

This PR does not contain connector or user-facing documentation changes and is outside the scope of the AI PR Review system.

Gate	Status
Out-of-Scope Changes	FAIL (PR is out of scope)
All other gates	UNKNOWN (not evaluated)

⚠️ Risk Level: 5/5

All functional changes are in the Kotlin bulk-load CDK core (dataflow/state/StateHistogramStore.kt, StateStore.kt, StateStage.kt, PipelineCompletionHandler.kt), which every bulk destination inherits on its next CDK bump — broad blast radius outside connector scope.

---

📋 PR Details

Connector(s): none (bulk CDK load module only; version bumped to 1.0.21)
PR: #83238
HEAD SHA: 4cac3d5f965e22fe0329e831e2f840c5574c68d1
Session: https://app.devin.ai/sessions/386ab9388612456c848767d95969f580

🔍 Scope Determination

Changed files analyzed: 11
In-scope files found: 0 connector files, 0 docs/** files

All 11 changed files live under airbyte-cdk/bulk/core/load/:

• src/main/kotlin/io/airbyte/cdk/load/dataflow/state/StateHistogramStore.kt
• src/main/kotlin/io/airbyte/cdk/load/dataflow/state/StateStore.kt
• src/main/kotlin/io/airbyte/cdk/load/dataflow/state/StateHistogram.kt
• src/main/kotlin/io/airbyte/cdk/load/dataflow/stages/StateStage.kt
• src/main/kotlin/io/airbyte/cdk/load/dataflow/pipeline/PipelineCompletionHandler.kt
• 4 corresponding test files under src/test/kotlin/...
• changelog.md, version.properties

The only markdown file is the CDK module's own changelog.md, which accompanies the code change rather than making this a documentation PR. The AI review gates are calibrated for changes under airbyte-integrations/connectors/** and docs/**; CDK core changes are explicitly out of scope and the Risk Level rubric assigns them 5/5, which forces the SKIP marker.

📚 Evidence Consulted

• Changed files: 11 (+304 / −91)
• PR labels: hyd-fix, hyd-prove, hyd-review
• PR description: present and detailed (root cause, scoping design, breaking-change evaluation, test coverage)
• Existing bot reviews: none for this HEAD SHA
• CI check-runs on HEAD SHA — 4 destination integration suites are currently failing, which is worth a human look since the PR description states these suites pass:
  • Test destination-databricks
  • Test destination-gcs-data-lake
  • Test destination-postgres
  • Test destination-redshift
  • Passing: destination-snowflake, destination-bigquery, destination-s3, destination-s3-data-lake, destination-clickhouse, destination-mssql, destination-azure-blob-storage, destination-dev-null, Bulk CDK version check, Check Changelog Updated, Format Check, Analyze Java/Kotlin
  • Also failing: Deploy Docs to Vercel (Preview)

These CI observations are reported as context only — no gate verdict was computed, because the PR is out of scope.

Recommended Action

Please request review from human maintainers who own the Kotlin bulk-load CDK (destinations platform team). Reviewers should pay particular attention to the four failing destination integration suites above and to the StateScope.Global branches that preserve cross-stream partition summing and removal.

No automated review action was taken.

[Airbyte Support Bot (airbyte-support-bot)]

🙋 Escalated to #human-in-the-loop per Hands-Free AI Triage Project triage next step.

Routed to the DB/DW connectors oncall. /ai-prove-fix is 🟢 Fix Proven on the current HEAD 4cac3d5f and the PR is marked ready, but /ai-review returned SKIP because all functional changes are in the Kotlin bulk-load CDK core (dataflow/state), which is out of scope for the automated connector review — every bulk destination inherits it on the next CDK bump. A human reviewer needs to take the review and merge decision.

---

Devin session