[tiering] Coordinator Should Push Lake Config to Tiering Service via Heartbeat RPC
- Dominant language
- Java
- Stars
- 2.1k
- Forks
- 625
- Avg merge
- 3d 14h
- Merged PRs (30d)
- 97
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
### Motivation
The current tiering service design requires the lake storage configuration (e.g., `--datalake.paimon.warehouse`) to be supplied **twice**:
1. **In `server.yaml`** — the Fluss coordinator uses it to create/manage the lake catalog (e.g., create Paimon table schema in the warehouse when `CREATE TABLE ... datalake.enabled=true` is called).
2. **In tiering CLI args** — the tiering service (Flink or Spark) uses the same values to initialize its `LakeTieringFactory` for writing lake files.
Both must point to the **same physical location**, but the system does not enforce this. A misconfigured tiering service silently writes to a wrong warehouse with no error.
This is a **single source of truth violation**: the coordinator already owns the authoritative lake config, yet forces the tiering service to redundantly re-specify it.
### Solution
Extend `LakeTieringHeartbeatResponse` → `PbLakeTieringTableInfo` to carry the lake format and config from the coordinator:
```proto
message PbLakeTieringTableInfo {
required int64 table_id = 1;
required PbTablePath table_path = 2;
required int64 tiering_epoch = 3;
required string data_lake_format = 4; // new
repeated PbKeyValuePair lake_config = 5; // new: e.g. warehouse=/path, metastore=filesystem
}
```
The coordinator already holds this information via `LakeCatalogContainer` (populated from `extractLakeProperties(conf)`). It would populate the new fields in `CoordinatorService.lakeTieringHeartbeat()`.
The tiering service would then:
- Remove `--datalake.format` and `--datalake..*` from its CLI contract.
- Lazily initialize `LakeTieringFactory` after receiving the first heartbeat response (instead of at startup).
- Embed the coordinator-supplied config into `TieringSplit` so Spark executors can reconstruct the factory without local CLI args.
## Benefits
1. **Eliminates misconfiguration risk** — warehouse path mismatch between coordinator and tiering service becomes impossible.
2. **Simpler deployment** — operators only configure lake storage once, in `server.yaml`.
3. **Enables multi-format tiering** — a single tiering service instance could eventually handle tables with different lake formats in one run, since the format is now per-table from the coordinator rather than a global CLI flag.
4. **Cleaner CLI** — `spark-submit ... SparkLakeTiering` only needs `--fluss.*` and `--lake.tiering.*` args; no lake-format-specific args.
### Anything else?
_No response_
### Willingness to contribute
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by tracing LakeTieringHeartbeatResponse and PbLakeTieringTableInfo into CoordinatorService.lakeTieringHeartbeat(), then inspect LakeCatalogContainer and LakeTieringFactory. Follow TieringSplit into the Spark tiering service and verify how the first heartbeat supplies configuration. Done means coordinator-supplied format and lake config reach executors without lake-format-specific CLI arguments.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- backend-api-design, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100