destination-gcs-data-lake: GCS writes fail after ~1 hour due to non-refreshable static OAuth2 token
- Lingua principale
- Python
- Stelle
- 22.1k
- Fork
- 5.3k
- Metriche di merge delle PR
- Metriche PR in attesa
Descrizione
### Connector Name
destination-gcs-data-lake
### Connector Version
1.0.9
### What step the error happened?
During the sync
### Relevant information
## Summary
The `destination-gcs-data-lake` connector fails with `OAuth2Credentials instance does not support refreshing the access token` after approximately 1 hour of sync time. This occurs because the connector extracts a static access token string from `ServiceAccountCredentials` and passes it to Iceberg's GCS FileIO, which cannot refresh it when it expires.
## Connector Version
- destination-gcs-data-lake: 1.0.7 (also confirmed on master/1.0.9)
## Error
```
java.lang.IllegalStateException: OAuth2Credentials instance does not support refreshing the access token. An instance with a new access token should be used, or a derived type that supports refreshing.
at com.google.auth.oauth2.OAuth2Credentials.refreshAccessToken(OAuth2Credentials.java:367)
at com.google.auth.oauth2.OAuth2Credentials$1.call(OAuth2Credentials.java:270)
...
at com.google.cloud.storage.spi.v1.HttpStorageRpc.open(HttpStorageRpc.java:1155)
at com.google.cloud.storage.ResumableMedia.lambda$startUploadForBlobInfo$0(ResumableMedia.java:38)
```
## Root Cause
In `GcsDataLakeCatalogUtil.toCatalogProperties()`:
```kotlin
val credentials = config.googleCredentials
credentials.refreshIfExpired()
val accessToken = credentials.accessToken.tokenValue
```
Then in `buildGcsProperties()`:
```kotlin
put(GCPProperties.GCS_OAUTH2_TOKEN, accessToken)
```
This passes a **static token string** to Iceberg. Iceberg's `GCSFileIO` wraps it in a plain `OAuth2Credentials.create(new AccessToken(...))` instance, which has no refresh capability. After ~1 hour (Google OAuth2 token default TTL), any GCS write attempt fails because the token is expired and cannot be refreshed.
## Steps to Reproduce
1. Configure `destination-gcs-data-lake` with a valid service account JSON and Polaris catalog type.
2. Start a sync that takes longer than 1 hour.
3. Observe failure at ~60 minutes with the above error.
## Expected Behavior
The connector should use a credential mechanism that supports token refresh for long-running syncs. GCS writes should not fail due to token expiration.
## Suggested Fix
Instead of extracting a static token and setting `GCPProperties.GCS_OAUTH2_TOKEN`, use one of:
1. **Write the service account JSON to a temp file and set `GCPProperties.GCS_OAUTH2_TOKEN_CREDENTIALS_FILE`** — Iceberg's `GCSFileIO` can then construct refreshable `ServiceAccountCredentials` internally.
2. **Set `GOOGLE_APPLICATION_CREDENTIALS` environment variable** — lets the Google auth library use ADC with full refresh support.
3. **Periodically re-create the catalog with a fresh token** — less ideal but a workaround.
## Environment
- Catalog type: Polaris (Apache Polaris 1.2.0-incubating)
- Storage: GCS
- Runtime: Kubernetes
- Sync duration: >1 hour
### Relevant log output
```shell
2026-05-22 13:39:00 destination ERROR main i.a.c.AirbyteConnectorRunnable(run):38 Failed class io.airbyte.cdk.load.write.WriteOperation operation execution. com.google.cloud.storage.StorageException: OAuth2Credentials instance does not support refreshing the access token. An instance with a new access token should be used, or a derived type that supports refreshing.
at com.google.cloud.storage.StorageException.getStorageException(StorageException.java:110) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at com.google.cloud.storage.StorageException.coalesce(StorageException.java:131) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at com.google.cloud.storage.Retrying$DefaultRetrier.run(Retrying.java:180) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at com.google.cloud.storage.Retrying$HttpRetrier.run(Retrying.java:206) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at com.google.cloud.storage.Retrying$RetrierWithAlgImpl.run(Retrying.java:137) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at com.google.cloud.storage.ResumableMedia.lambda$startUploadForBlobInfo$1(ResumableMedia.java:34) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at com.google.cloud.storage.StorageImpl.writer(StorageImpl.java:765) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at com.google.cloud.storage.StorageImpl.writer(StorageImpl.java:101) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at org.apache.iceberg.gcp.gcs.GCSOutputStream.openStream(GCSOutputStream.java:114) ~[iceberg-gcp-1.7.0.jar:?]
at org.apache.iceberg.gcp.gcs.GCSOutputStream.(GCSOutputStream.java:74) ~[iceberg-gcp-1.7.0.jar:?]
at org.apache.iceberg.gcp.gcs.GCSOutputFile.createOrOverwrite(GCSOutputFile.java:62) ~[iceberg-gcp-1.7.0.jar:?]
at org.apache.iceberg.parquet.ParquetIO$ParquetOutputFile.createOrOverwrite(ParquetIO.java:151) ~[iceberg-parquet-1.7.0.jar:?]
at org.apache.parquet.hadoop.ParquetFileWriter.(ParquetFileWriter.java:345) ~[parquet-hadoop-1.13.1.jar:1.13.1]
at org.apache.parquet.hadoop.ParquetFileWriter.(ParquetFileWriter.java:323) ~[parquet-hadoop-1.13.1.jar:1.13.1]
at org.apache.iceberg.parquet.ParquetWriter.ensureWriterInitialized(ParquetWriter.java:111) ~[iceberg-parquet-1.7.0.jar:?]
at org.apache.iceberg.parquet.ParquetWriter.flushRowGroup(ParquetWriter.java:210) ~[iceberg-parquet-1.7.0.jar:?]
at org.apache.iceberg.parquet.ParquetWriter.close(ParquetWriter.java:254) ~[iceberg-parquet-1.7.0.jar:?]
at org.apache.iceberg.io.DataWriter.close(DataWriter.java:82) ~[iceberg-core-1.7.0.jar:?]
at org.apache.iceberg.io.BaseTaskWriter$BaseRollingWriter.closeCurrent(BaseTaskWriter.java:344) ~[iceberg-core-1.7.0.jar:?]
at org.apache.iceberg.io.BaseTaskWriter$BaseRollingWriter.close(BaseTaskWriter.java:371) ~[iceberg-core-1.7.0.jar:?]
at org.apache.iceberg.io.UnpartitionedWriter.close(UnpartitionedWriter.java:47) ~[iceberg-core-1.7.0.jar:?]
at org.apache.iceberg.io.BaseTaskWriter.complete(BaseTaskWriter.java:99) ~[iceberg-core-1.7.0.jar:?]
at io.airbyte.integrations.destination.gcs_data_lake.dataflow.GcsDataLakeAggregate.flush(GcsDataLakeAggregate.kt:110) ~[io.airbyte.airbyte-integrations.connectors-destination-gcs-data-lake.jar:?]
at io.airbyte.cdk.load.dataflow.stages.FlushStage.apply(FlushStage.kt:17) ~[bulk-cdk-core-load-1.0.9.jar:?]
at io.airbyte.cdk.load.dataflow.pipeline.DataFlowPipeline$run$$inlined$map$2$2.emit(Emitters.kt:220) ~[bulk-cdk-core-load-1.0.9.jar:?]
at kotlinx.coroutines.flow.FlowKt__ChannelsKt.emitAllImpl$FlowKt__ChannelsKt(Channels.kt:33) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
at kotlinx.coroutines.flow.FlowKt__ChannelsKt.access$emitAllImpl$FlowKt__ChannelsKt(Channels.kt:1) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
at kotlinx.coroutines.flow.FlowKt__ChannelsKt$emitAllImpl$1.invokeSuspend(Channels.kt) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33) ~[kotlin-stdlib-2.1.10.jar:2.1.10-release-473]
at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:104) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:111) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:99) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:584) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:811) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:715) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:702) ~[kotlinx-coroutines-core-jvm-1.8.1.jar:?]
Suppressed: com.google.cloud.storage.RetryContext$RetryBudgetExhaustedComment: Unretryable error (attempts: 1, maxAttempts: 6, elapsed: PT0.04S, nextBackoff: PT1.296737641S, timeout: PT50S)
Caused by: java.lang.IllegalStateException: OAuth2Credentials instance does not support refreshing the access token. An instance with a new access token should be used, or a derived type that supports refreshing.
at com.google.auth.oauth2.OAuth2Credentials.refreshAccessToken(OAuth2Credentials.java:367) ~[google-auth-library-oauth2-http-1.40.0.jar:1.40.0]
at com.google.auth.oauth2.OAuth2Credentials$1.call(OAuth2Credentials.java:270) ~[google-auth-library-oauth2-http-1.40.0.jar:1.40.0]
at com.google.auth.oauth2.OAuth2Credentials$1.call(OAuth2Credentials.java:267) ~[google-auth-library-oauth2-http-1.40.0.jar:1.40.0]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317) ~[?:?]
at com.google.auth.oauth2.OAuth2Credentials$RefreshTask.run(OAuth2Credentials.java:644) ~[google-auth-library-oauth2-http-1.40.0.jar:1.40.0]
at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30) ~[guava-33.5.0-jre.jar:?]
at com.google.auth.oauth2.OAuth2Credentials$AsyncRefreshResult.executeIfNew(OAuth2Credentials.java:591) ~[google-auth-library-oauth2-http-1.40.0.jar:1.40.0]
at com.google.auth.oauth2.OAuth2Credentials.refresh(OAuth2Credentials.java:193) ~[google-auth-library-oauth2-http-1.40.0.jar:1.40.0]
at com.google.auth.http.HttpCredentialsAdapter.handleResponse(HttpCredentialsAdapter.java:146) ~[google-auth-library-oauth2-http-1.40.0.jar:1.40.0]
at com.google.api.client.http.HttpRequest.execute(HttpRequest.java:1059) ~[google-http-client-2.0.2.jar:2.0.2]
at com.google.cloud.storage.spi.v1.HttpStorageRpc.open(HttpStorageRpc.java:1155) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at com.google.cloud.storage.ResumableMedia.lambda$startUploadForBlobInfo$0(ResumableMedia.java:38) ~[google-cloud-storage-2.59.0.jar:2.59.0]
at com.google.cloud.storage.Retrying$DefaultRetrier.run(Retrying.java:165) ~[google-cloud-storage-2.59.0.jar:2.59.0]
... 33 more
Stack Trace: com.google.cloud.storage.StorageException: OAuth2Credentials instance does not support refreshing the access token. An instance with a new access token should be used, or a derived type that supports refreshing.
at com.google.cloud.storage.StorageException.getStorageException(StorageException.java:110)
at com.google.cloud.storage.StorageException.coalesce(StorageException.java:131)
at com.google.cloud.storage.Retrying$DefaultRetrier.run(Retrying.java:180)
at com.google.cloud.storage.Retrying$HttpRetrier.run(Retrying.java:206)
at com.google.cloud.storage.Retrying$RetrierWithAlgImpl.run(Retrying.java:137)
at com.google.cloud.storage.ResumableMedia.lambda$startUploadForBlobInfo$1(ResumableMedia.java:34)
at com.google.cloud.storage.StorageImpl.writer(StorageImpl.java:765)
at com.google.cloud.storage.StorageImpl.writer(StorageImpl.java:101)
at org.apache.iceberg.gcp.gcs.GCSOutputStream.openStream(GCSOutputStream.java:114)
at org.apache.iceberg.gcp.gcs.GCSOutputStream.(GCSOutputStream.java:74)
at org.apache.iceberg.gcp.gcs.GCSOutputFile.createOrOverwrite(GCSOutputFile.java:62)
at org.apache.iceberg.parquet.ParquetIO$ParquetOutputFile.createOrOverwrite(ParquetIO.java:151)
at org.apache.parquet.hadoop.ParquetFileWriter.(ParquetFileWriter.java:345)
at org.apache.parquet.hadoop.ParquetFileWriter.(ParquetFileWriter.java:323)
at org.apache.iceberg.parquet.ParquetWriter.ensureWriterInitialized(ParquetWriter.java:111)
at org.apache.iceberg.parquet.ParquetWriter.flushRowGroup(ParquetWriter.java:210)
at org.apache.iceberg.parquet.ParquetWriter.close(ParquetWriter.java:254)
at org.apache.iceberg.io.DataWriter.close(DataWriter.java:82)
at org.apache.iceberg.io.BaseTaskWriter$BaseRollingWriter.closeCurrent(BaseTaskWriter.java:344)
at org.apache.iceberg.io.BaseTaskWriter$BaseRollingWriter.close(BaseTaskWriter.java:371)
at org.apache.iceberg.io.UnpartitionedWriter.close(UnpartitionedWriter.java:47)
at org.apache.iceberg.io.BaseTaskWriter.complete(BaseTaskWriter.java:99)
at io.airbyte.integrations.destination.gcs_data_lake.dataflow.GcsDataLakeAggregate.flush(GcsDataLakeAggregate.kt:110)
at io.airbyte.cdk.load.dataflow.stages.FlushStage.apply(FlushStage.kt:17)
at io.airbyte.cdk.load.dataflow.pipeline.DataFlowPipeline$run$$inlined$map$2$2.emit(Emitters.kt:220)
at kotlinx.coroutines.flow.FlowKt__ChannelsKt.emitAllImpl$FlowKt__ChannelsKt(Channels.kt:33)
at kotlinx.coroutines.flow.FlowKt__ChannelsKt.access$emitAllImpl$FlowKt__ChannelsKt(Channels.kt:1)
at kotlinx.coroutines.flow.FlowKt__ChannelsKt$emitAllImpl$1.invokeSuspend(Channels.kt)
at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:104)
at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:111)
at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:99)
at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:584)
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:811)
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:715)
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:702)
Suppressed: com.google.cloud.storage.RetryContext$RetryBudgetExhaustedComment: Unretryable error (attempts: 1, maxAttempts: 6, elapsed: PT0.04S, nextBackoff: PT1.296737641S, timeout: PT50S)
Caused by: java.lang.IllegalStateException: OAuth2Credentials instance does not support refreshing the access token. An instance with a new access token should be used, or a derived type that supports refreshing.
at com.google.auth.oauth2.OAuth2Credentials.refreshAccessToken(OAuth2Credentials.java:367)
at com.google.auth.oauth2.OAuth2Credentials$1.call(OAuth2Credentials.java:270)
at com.google.auth.oauth2.OAuth2Credentials$1.call(OAuth2Credentials.java:267)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
at com.google.auth.oauth2.OAuth2Credentials$RefreshTask.run(OAuth2Credentials.java:644)
at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
at com.google.auth.oauth2.OAuth2Credentials$AsyncRefreshResult.executeIfNew(OAuth2Credentials.java:591)
at com.google.auth.oauth2.OAuth2Credentials.refresh(OAuth2Credentials.java:193)
at com.google.auth.http.HttpCredentialsAdapter.handleResponse(HttpCredentialsAdapter.java:146)
at com.google.api.client.http.HttpRequest.execute(HttpRequest.java:1059)
at com.google.cloud.storage.spi.v1.HttpStorageRpc.open(HttpStorageRpc.java:1155)
at com.google.cloud.storage.ResumableMedia.lambda$startUploadForBlobInfo$0(ResumableMedia.java:38)
at com.google.cloud.storage.Retrying$DefaultRetrier.run(Retrying.java:165)
... 33 more
```
### Contribute
- [ ] Yes, I want to contribute
---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/12691
Guida per i contributori
Apri la guida per i contributori
Valutazione
Questa issue non è ancora stata valutata.