airbytehq / airbytehq/airbyte

[destination-gcs-data-lake] Deterministic Iceberg CommitFailedException on `airbyte_staging` under high source parallelism (SOCKET, max_db_connections=8)

Đang mở
#80,936 3 bình luận 0 reaction 0 người được giao Xem trên GitHub
autoteam community connectors/destination/gcs-data-lake connectors/source/postgres hyd-review team/use
Ngôn ngữ chính
Python
Star
22.1k
Fork
5.3k
Chỉ số merge pull request
Chỉ số pull request đang chờ

Mô tả

## Connector Name

destination-gcs-data-lake

## Connector Version

1.0.10 (Airbyte 2.1.0 OSS)

## What step the error happened?

During the sync (destination write / Iceberg commit)

## Relevant information

Full-refresh sync of a single large PostgreSQL table (~32M rows / ~12 GB) into a **BigLake Iceberg REST catalog** on GKE Autopilot. Source is `source-postgres` 3.8.1 using the **SOCKET** data channel (Speed Mode, PROTOBUF); `destination-gcs-data-lake` 1.0.10 also negotiates SOCKET.

The failure is **parallelism-dependent and deterministic**:

- Source `max_db_connections = 4` → sync **succeeds reliably**, data complete (records committed == emitted).
- Source `max_db_connections = 8` → the destination fails **deterministically** — on both the initial attempt **and** Airbyte's automatic retry — with an Iceberg optimistic-concurrency conflict on the shared `airbyte_staging` branch. The source then dies with `Broken pipe` (its SOCKET peer has exited).

Mechanism: raising `max_db_connections` increases the source emit rate, so the destination flushes and **commits Iceberg snapshots to the single `airbyte_staging` branch more frequently / concurrently**. Concurrent commits collide on `assert-ref-snapshot-id`, and the connector surfaces the conflict as **fatal** instead of retrying or serializing the commit. At 4 readers the commit rate stays low enough to avoid the collision.

This is a single sync of a single stream, so the collision is **intra-sync** (concurrent flush pipelines within one sync), not cross-sync.

### Related work on `destination-s3-data-lake` (shared Iceberg / Bulk-CDK toolkit)

- #79112 — *use unique staging branches* (merged). Addresses shared-`airbyte_staging` collisions, but it is framed as **cross-sync**; a per-(stream, generation) branch may not cover concurrent flush pipelines **within a single stream's sync**.
- #76348 / #76343 — *increase aggregate flush size / small files*. Fewer, larger commits would reduce the commit-collision pressure.

Neither change appears to be present in `destination-gcs-data-lake` 1.0.10. Could these be ported to gcs-data-lake, and/or could the Iceberg commit path get a **retry-with-backoff** (or commit serialization) so that high source parallelism doesn't fail the write?

## Relevant log output

Destination (root cause, abridged — snapshot ids redacted):

```
ERROR i.a.c.l.d.p.PipelineRunner Destination Pipeline Completed — Exceptionally
org.apache.iceberg.exceptions.CommitFailedException: Commit failed: Conflict updating the table. Please retry.

org.apache.iceberg.exceptions.CommitFailedException: Commit failed: Requirement failed:
branch airbyte_staging has changed: expected id != .; Failed to assert requirement type assert-ref-snapshot-id
at org.apache.iceberg.rest.ErrorHandlers$CommitErrorHandler.accept(ErrorHandlers.java:89)
at org.apache.iceberg.rest.RESTTableOperations.commit(RESTTableOperations.java:159)
at org.apache.iceberg.SnapshotProducer.commit(SnapshotProducer.java:401)
at io.airbyte.integrations.destination.gcs_data_lake.dataflow.GcsDataLakeAggregate.flush(GcsDataLakeAggregate.kt:122)
at io.airbyte.cdk.load.dataflow.stages.FlushStage.apply(FlushStage.kt:17)
...
```

Source (consequence — its socket peer is gone):

```
ERROR i.a.c.o.w.i.LocalContainerAirbyteSource java.io.IOException: Broken pipe
at io.airbyte.cdk.output.sockets.SocketProtobufOutputConsumer.withLockFlush(SocketProtobufOutputConsumer.kt:69)
...
```

## Expected behavior

Under high source parallelism the destination should tolerate concurrent Iceberg commits to its staging branch (retry-with-backoff, commit serialization, or a per-writer unique branch) rather than failing the whole sync.

## Environment

- Airbyte 2.1.0 (OSS, Helm chart 2.1.0), GKE Autopilot
- `source-postgres` 3.8.1 — SOCKET / PROTOBUF (Speed Mode), full-refresh
- `destination-gcs-data-lake` 1.0.10 — BigLake Iceberg REST catalog

## Are you willing to submit a PR?

No

---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/13009

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.