prashantwason commented on code in PR #18411:
URL: https://github.com/apache/hudi/pull/18411#discussion_r3011359344
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -228,14 +228,18 @@ public void start() throws Exception {
this.metaClient = initTableIfNotExists(this.conf);
// the write client must create after the table creation
this.writeClient = FlinkWriteClients.createWriteClient(conf);
- this.writeClient.tryUpgrade(instant, this.metaClient);
- initMetadataTable(this.writeClient);
- if (tableState.scheduleMdtCompaction) {
- this.metadataWriteClient =
StreamerUtil.createMetadataWriteClient(writeClient);
- }
-
- // start the executor
+ // Create executors BEFORE the upgrade so that the heavy initialization
+ // (upgrade, metadata table init, event restoration) can run on the
executor
+ // thread instead of the Pekko dispatcher thread. Running the upgrade
+ // synchronously on the dispatcher thread blocks heartbeat responses and
+ // causes the ResourceManager to disconnect the JobManager when upgrades
+ // take longer than the heartbeat timeout (e.g., LSM timeline migration
+ // with hundreds of archived actions).
+ //
+ // Since the executor is single-threaded and FIFO, submitting the
Review Comment:
Good points — the race between `startInstant()` on `instantRequestExecutor`
and the async upgrade on `executor` is a real issue, and the RLI dependency on
MDT init makes it unsafe to start the pipeline early.
Reverted the async change completely. The revised commit only optimizes
`upgradeToLSMTimeline()` to reduce the I/O cost of the migration itself (batch
size 500 + single compactAndClean at end), keeping the synchronous start() flow
intact.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
