[Destination S3 Data Lake] Glue Catalog destination OOMKilled — Possible Memory Leak in S3FileIO Buffering

### Connector Name

S3 Data lake

### Connector Version

0.3.41

### What step the error happened?

During the sync

### Relevant information

### **Issue**

* `destination` container exits due to **OOMKilled (137)**
* `source` and `orchestrator` containers exit cleanly
* Logs repeatedly show unclosed S3 file warnings:

```
WARN Finalizer o.a.i.a.s.S3FileIO(finalize):431 Unclosed S3FileIO instance created by:
org.apache.iceberg.aws.s3.S3FileIO.initialize(...)
```

---

### **Impact**

* Sync stops after ~5–7 minutes
* Only ~6.5M rows / ~2GB finish writing
* Full 50GB load becomes impossible to run

---

### **Kubernetes Pod Resource Configuration**

```
source: 20Gi limit / 8Gi request
destination: 20Gi limit / 8Gi request <-- OOMKilled here
orchestrator: 20Gi limit / 8Gi request
```

### **Cluster Node**

```
RAM: ~32GB
```

---

### **Dataset Characteristics**

```
MySQL RDS dataset: ~50GB
Total rows: ~100M+
Rows processed before crash: ~6.5M (~2GB of written data)
Sync mode: Incremental Sync + dedup → Iceberg S3 Data Lake
Catalog: AWS Glue
```

---

### **Environment**

```
Airbyte OSS: 2.0.0
source-mysql: 3.51.1
destination-s3-data-lake (Iceberg): 0.3.41
```

---

### **Hypothesis**

* Iceberg AWS `S3FileIO` objects are not being closed properly
* Destination is buffering files in-memory instead of streaming
* File transfer / staging settings ignored internally

---

### **Steps to Reproduce**

1. Deploy Airbyte OSS 2.0.0 to Kubernetes
2. Configure **MySQL → S3 Data Lake (Iceberg + Glue Catalog)** connection
3. Run full-refresh sync of 50GB+ database
4. Observe memory usage (`destination` climbs until OOM)

---

### **Exit Codes**

```
destination: 137 (OOMKilled)
source: 143 (terminated after destination died)
orchestrator: 0 (clean shutdown)
```

---

## **Request**

Please investigate:

* Potential memory leak in Iceberg `S3FileIO` or write buffering pipeline
* Whether file chunking / streaming is working as intended
* Whether `USE_FILE_TRANSFER` or `DATA_CHANNEL_MEDIUM=FILE` is supported.

### Relevant log output

```shell
2025-11-28 10:54:41,655 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account-round-1-partition-1-read i.a.c.r.JdbcSelectQuerier$Result(initQueryExecution):95 Setting Statement fetchSize to -2147483648.
2025-11-28 10:54:42,212 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account-round-1-partition-1-read i.a.c.r.JdbcSelectQuerier$Result(initQueryExecution):106 Setting ResultSet fetchSize to 32421425.
2025-11-28 10:54:42,214 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account-round-1-partition-1-read i.a.c.r.JdbcSelectQuerier$Result(hasNext):117 Received results from server.
2025-11-28 10:54:42,265 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4 i.a.c.l.s.PipelineEventBookkeepingRouter(handleStreamMessage):101 Saw first record for stream Descriptor(namespace=musoni, name=acc_gl_account); awaiting setup complete
2025-11-28 10:54:42,266 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4 i.a.c.l.s.PipelineEventBookkeepingRouter(handleStreamMessage):105 Setup complete, starting stream Descriptor(namespace=musoni, name=acc_gl_account)
2025-11-28 10:54:42,272 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-6 i.a.c.l.t.i.UpdateCheckpointsTask$execute$3(emit):57 Updating global checkpoint with CheckpointKey(checkpointIndex=CheckpointIndex(value=6), checkpointId=CheckpointId(value=6)):Stats(recordCount=1956660, rejectedRecordCount=0, additionalStats={})
2025-11-28 10:54:42,276 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-6 i.a.c.l.s.CheckpointManager(addGlobalCheckpoint):135 Added global checkpoint with key CheckpointKey(checkpointIndex=CheckpointIndex(value=6), checkpointId=CheckpointId(value=6))
2025-11-28 10:54:42,409 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4 o.a.i.BaseMetastoreTableOperations(refreshFromMetadataLocation):189 Refreshing table metadata from new version: s3://fermi-staging-airbyte-sync/prabal-org/musoni.db/acc_gl_account/metadata/00001-d3d49982-7e5f-42c2-9be5-10aad22e8da0.metadata.json
2025-11-28 10:54:42,450 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account-round-1-partition-1-read i.a.c.r.JdbcSelectQuerier$Result(close):173 Closing SELECT `id`, `name`, `tag_id`, `gl_code`, `disabled`, `hierarchy`, `parent_id`, `description`, `external_id`, `account_usage`, `classification_enum`, `last_entry_id_derived`, `reconciliation_enabled`, `manual_journal_entries_allowed`, `organization_running_balance_derived` FROM `mifostenant-agora`.`acc_gl_account`
2025-11-28 10:54:42,508 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4 o.a.i.BaseMetastoreCatalog(loadTable):69 Table loaded by catalog: airbyte.musoni.acc_gl_account
2025-11-28 10:54:42,508 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4 i.a.c.l.t.i.p.i.IcebergUtil(createTable):145 Loading Iceberg table musoni.acc_gl_account ...
2025-11-28 10:54:42,637 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account-round-1-partition-1-read i.a.c.r.FeedReader(readPartitionWithResources):239 completed reading partition 1 for 'mifostenant-agora.acc_gl_account' in round 1
2025-11-28 10:54:42,637 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account-round-1-partition-1-read i.a.c.r.FeedReader(readPartitionWithResources):251 releasing resources acquired to read partition 1 for 'mifostenant-agora.acc_gl_account' in round 1
2025-11-28 10:54:42,637 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account-round-1-partition-1-read i.a.c.r.FeedReader(readPartitionWithResources):258 read 870 record(s) from partition 1 for 'mifostenant-agora.acc_gl_account' in round 1
2025-11-28 10:54:42,637 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account i.a.c.r.FeedReader(awaitAllPartitionReaders):316 processing result (success = true) from reading partition 1 / 1 for 'mifostenant-agora.acc_gl_account'
2025-11-28 10:54:42,638 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account i.a.c.r.FeedReader(awaitAllPartitionReaders):336 updated state of 'mifostenant-agora.acc_gl_account', moved it 870 record(s) forward
2025-11-28 10:54:42,638 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account i.a.c.r.FeedReader(maybeCheckpoint):387 checkpoint of 1 state message(s)
2025-11-28 10:54:42,641 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account i.a.c.r.FeedReader(createPartitions):107 Attempting bootstrap using class io.airbyte.cdk.read.cdc.CdcPartitionsCreatorFactory.
2025-11-28 10:54:42,641 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account i.a.c.r.FeedReader(createPartitions):107 Attempting bootstrap using class io.airbyte.integrations.source.mysql.MySqlJdbcConcurrentPartitionsCreatorFactory.
2025-11-28 10:54:42,642 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-3#mifostenant-agora.acc_gl_account-round-2-acquire-resources i.a.c.r.FeedReader(acquirePartitionsCreatorResources):136 acquired resources to create partitions for 'mifostenant-agora.acc_gl_account' in round 2
2025-11-28 10:54:42,642 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-7#mifostenant-agora.acc_gl_account-round-2-create-partitions i.a.c.r.FeedReader(createPartitionsWithResources):146 creating partitions for 'mifostenant-agora.acc_gl_account' in round 2
2025-11-28 10:54:42,643 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-7#mifostenant-agora.acc_gl_account-round-2-create-partitions i.a.c.r.FeedReader(createPartitionsWithResources):150 releasing resources acquired to create partitions for 'mifostenant-agora.acc_gl_account' in round 2
2025-11-28 10:54:42,644 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-7#mifostenant-agora.acc_gl_account i.a.c.r.FeedReader(read):72 no more partitions to read for 'mifostenant-agora.acc_gl_account' in round 2
2025-11-28 10:54:42,645 [pool-4-thread-3] INFO i.a.c.o.b.s.StreamStatusTracker(trackEvent):108 - Stream status TRACE received of status: COMPLETE for stream mifostenant-agora:acc_gl_account
2025-11-28 10:54:42,648 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-12#mifostenant-agora.acc_accounting_rule_entry_mapping-round-1-acquire-resources i.a.c.r.FeedReader(acquirePartitionsCreatorResources):136 acquired resources to create partitions for 'mifostenant-agora.acc_accounting_rule_entry_mapping' in round 1
2025-11-28 10:54:42,649 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-12#mifostenant-agora.acc_accounting_rule_entry_mapping-round-1-create-partitions i.a.c.r.FeedReader(createPartitionsWithResources):146 creating partitions for 'mifostenant-agora.acc_accounting_rule_entry_mapping' in round 1
2025-11-28 10:54:42,650 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-12#mifostenant-agora.acc_accounting_rule_entry_mapping-round-1-create-partitions i.a.c.r.JdbcPartitionsCreator(collectSample):99 Sampling stream 'mifostenant-agora.acc_accounting_rule_entry_mapping' at rate 1 / 65536.
2025-11-28 10:54:42,650 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-12#mifostenant-agora.acc_accounting_rule_entry_mapping-round-1-create-partitions i.a.c.r.JdbcSelectQuerier$Result():76 Querying SELECT `id`, `debit_office_id`, `credit_office_id`, `debit_account_id`, `credit_account_id`, `acc_accounting_rule_id` FROM `mifostenant-agora`.`acc_accounting_rule_entry_mapping` ORDER BY `id` LIMIT ?
2025-11-28 10:54:42,651 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-12#mifostenant-agora.acc_accounting_rule_entry_mapping-round-1-create-partitions i.a.c.j.JdbcConnectionFactory(get):44 Creating new connection for 'jdbc:mysql://musonidb.c632usuiuw3l.us-east-1.rds.amazonaws.com:3306'.
2025-11-28 10:54:42,672 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4#read i.a.c.r.ReadOperation$execute$1$1$1(invokeSuspend):80 coroutine state:
read
├─mifostenant-agora.acc_accounting_rule_entry_mapping
│ └─mifostenant-agora.acc_accounting_rule_entry_mapping-round-1-create-partitions
└─mifostenant-agora.acc_gl_daily_movements
└─mifostenant-agora.acc_gl_daily_movements-round-1-acquire-resources

2025-11-28 10:54:42,717 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4 o.a.i.BaseMetastoreTableOperations(refreshFromMetadataLocation):189 Refreshing table metadata from new version: s3://fermi-staging-airbyte-sync/prabal-org/musoni.db/acc_gl_account/metadata/00001-d3d49982-7e5f-42c2-9be5-10aad22e8da0.metadata.json
2025-11-28 10:54:42,783 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4 o.a.i.BaseMetastoreCatalog(loadTable):69 Table loaded by catalog: airbyte.musoni.acc_gl_account
2025-11-28 10:54:42,786 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4 i.a.i.d.s.S3DataLakeStreamLoader(start):74 maybe creating branch airbyte_staging for stream Descriptor(namespace=musoni, name=acc_gl_account)
2025-11-28 10:54:42,859 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-4 i.a.i.d.s.S3DataLakeStreamLoader(start):79 branch airbyte_staging already exists for stream Descriptor(namespace=musoni, name=acc_gl_account)
2025-11-28 10:54:42,860 [pool-4-thread-4] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-1 i.a.c.l.s.PipelineEventBookkeepingRouter(handleStreamMessage):108 Initialization for stream Descriptor(namespace=musoni, name=acc_gl_account) complete
2025-11-28 10:54:44,051 [pool-4-thread-4] WARN i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):245 - WARN Finalizer o.a.i.a.s.S3FileIO(finalize):431 Unclosed S3FileIO instance created by:
org.apache.iceberg.aws.s3.S3FileIO.initialize(S3FileIO.java:371)
org.apache.iceberg.aws.glue.GlueTableOperations.initializeFileIO(GlueTableOperations.java:220)
org.apache.iceberg.aws.glue.GlueTableOperations.io(GlueTableOperations.java:115)
org.apache.iceberg.io.FileIOTracker.track(FileIOTracker.java:52)
org.apache.iceberg.aws.glue.GlueCatalog.newTableOps(GlueCatalog.java:244)
org.apache.iceberg.BaseMetastoreCatalog.loadTable(BaseMetastoreCatalog.java:48)
org.apache.iceberg.catalog.Catalog.tableExists(Catalog.java:281)
io.airbyte.cdk.load.toolkits.iceberg.parquet.io.IcebergUtil.createTable(IcebergUtil.kt:137)
io.airbyte.integrations.destination.s3_data_lake.S3DataLakeStreamLoader.start(S3DataLakeStreamLoader.kt:56)
io.airbyte.cdk.load.task.implementor.OpenStreamTask$execute$2.emit(OpenStreamTask.kt:43)
io.airbyte.cdk.load.task.implementor.OpenStreamTask$execute$2.emit(OpenStreamTask.kt:36)
kotlinx.coroutines.flow.FlowKt__ChannelsKt.emitAllImpl$FlowKt__ChannelsKt(Channels.kt:33)
kotlinx.coroutines.flow.FlowKt__ChannelsKt.access$emitAllImpl$FlowKt__ChannelsKt(Channels.kt:1)
kotlinx.coroutines.flow.FlowKt__ChannelsKt$emitAllImpl$1.invokeSuspend(Channels.kt)
kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:100)
kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:124)
kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:89)
kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:586)
kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:820)
kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:717)
kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:704)
2025-11-28 10:54:44,882 [pool-4-thread-2] INFO i.a.w.i.VersionedAirbyteStreamFactory(internalLog$io_airbyte_airbyte_commons_worker):248 - INFO DefaultDispatcher-worker-12#mifostenant-agora.acc_accounting_rule_entry_mapping-round-1-create-partitions i.a.c.r.JdbcSelectQuerier$Result(initQueryExecution):95 Setting Statement fetchSize to -2147483648.
```

### Contribute

- [ ] Yes, I want to contribute