Flink streaming job - token expiration
- Dominant language
- Java
- Stars
- 2.1k
- Forks
- 522
- Avg merge
- 1d 22h
- Merged PRs (30d)
- 137
Description
### Describe the bug
Experiencing authentication failures due to expired tokens when streaming from Kafka to Iceberg via PyFlink, running on Azure Blob Storage with Polaris catalog.
- Flink Version: 1.18
- Iceberg Version: 1.9.0
- Polaris Version: 0.11.0
- Storage: Azure Blob Storage
- Streaming Framework: PyFlink
The streaming job runs successfully for approximately 1 hour, then fails with an authentication error indicating expired tokens. The job needs to be manually restarted to continue processing.
```
AuthenticationFailedServer failed to authenticate the request. Make sure the value of Authorization header is formed correctly including the signature.
Time:2025-04-29T10:04:32.3424886ZSigned expiry time [Tue, 29 Apr 2025 09:39:38 GMT] must be after signed start time [Tue, 29 Apr 2025 10:04:32 GMT]
```
Polaris-side logs showing successful reauthentication:
```
2025-04-29 09:31:31,217 INFO [io.qua.htt.access-log] [,POLARIS] [,,,] (executor-thread-1) 10.244.4.122 - - [29/April/2025:09:31:31 +0000] "POST /api/catalog/v1/oauth/tokens HTTP/1.1" 200 770
2025-04-29 09:31:33,661 INFO [com.azu.ide.ChainedTokenCredential] [,] [,,,] (azure-sdk-global-thread-1) Azure Identity => Attempted credential EnvironmentCredential returns a token
```
### To Reproduce
_No response_
### Actual Behavior
The job fails after ~1 hour with authentication errors and requires manual restart to resume processing.
### Expected Behavior
The streaming job should automatically refresh authentication tokens before they expire, allowing for continuous operation without manual intervention.
### Additional context
The catalog is created using Flink SQL
```
tbl_env.execute_sql(f"""
CREATE CATALOG {catalog_name} WITH (
'type' = 'iceberg',
'catalog-type' = 'rest',
'uri' = '{catalog_uri}',
'credential' = '{client_id}:{client_secret}',
'scope' = 'PRINCIPAL_ROLE:ALL',
'warehouse' = '{warehouse}',
'token-refresh-enabled' = 'true',
'header.Polaris-realm' = 'POLARIS',
'header.X-Iceberg-Access-Delegation' = 'vended-credentials',
'io-impl' = 'org.apache.iceberg.io.ResolvingFileIO'
);
""")
```
### System information
_No response_
Contributor guide
Research direction
Start with the Flink SQL CREATE CATALOG configuration, especially token-refresh-enabled, and compare it with the Polaris token and Azure Identity log entries around the failure. Trace how authentication is refreshed during the long-running Kafka-to-Iceberg stream. Done means the job continues beyond token expiry without a manual restart and the expired-token failure no longer occurs.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- azure, java, python
- Domain
- authentication, cloud, data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 30/100