[Bug] Job cannot recover from checkpoint/savepoint if parallelism is changed from 1 to 2 #4543

Activity

1. Tan-JiaLiang

   added

   bugSomething isn't workingSomething isn't working

   on Nov 18, 2024
2. Tan-JiaLiang commented on Nov 18, 2024

   

   Tan-JiaLiang

   on Nov 18, 2024

   ContributorAuthor

   More actions

   paimon/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketSink.java

   Lines 69 to 83 in 220789d

   	if (enableCompaction && isStreamingMode) {
   	written =
   	written.transform(
   	"Compact Coordinator: " + table.name(),
   	new EitherTypeInfo<>(
   	new CommittableTypeInfo(),
   	new CompactionTaskTypeInfo()),
   	new AppendBypassCoordinateOperatorFactory<>(table))
   	.forceNonParallel()
   	.transform(
   	"Compact Worker: " + table.name(),
   	new CommittableTypeInfo(),
   	new AppendBypassCompactWorkerOperator(table, initialCommitUser))
   	.setParallelism(written.getParallelism());
   	}

   If the job parallelism is 1, the Writer operator and the Compact Coordinator operator will be chained. However, since the parallelism of the Compact Coordinator operator is always 1, when the job parallelism is adjusted, the Writer operator and the Compact Coordinator operator will be separated, resulting in the state not being recoverable.

   We need to disable chain between writer operator and compact corrdinator operator.
3. Tan-JiaLiang commented on Nov 18, 2024

   

   Tan-JiaLiang

   on Nov 18, 2024

   ContributorAuthor

   More actions

   paimon/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java

   Lines 243 to 256 in 220789d

   	if (options.get(CHANGELOG_PRECOMMIT_COMPACT)) {
   	written =
   	written.transform(
   	"Changelog Compact Coordinator",
   	new EitherTypeInfo<>(
   	new CommittableTypeInfo(), new ChangelogTaskTypeInfo()),
   	new ChangelogCompactCoordinateOperator(table))
   	.forceNonParallel()
   	.transform(
   	"Changelog Compact Worker",
   	new CommittableTypeInfo(),
   	new ChangelogCompactWorkerOperator(table))
   	.setParallelism(written.getParallelism());
   	}

   Same as org.apache.paimon.flink.sink.FlinkSink#doWrite
4. Tan-JiaLiang commented on Nov 18, 2024

   

   Tan-JiaLiang

   on Nov 18, 2024

   ContributorAuthor

   More actions

   @JingsongLi WDYT? Should we add an option to control this? Like #3232.
5. Tan-JiaLiang commented on Nov 19, 2024

   

   Tan-JiaLiang

   on Nov 19, 2024

   ContributorAuthor

   More actions

   I think #4424 can solve this problem. But do we still need to add the disable chain? Or do we just need to recommend that users add the 'sink.operator-uid.suffix' and 'source.operator-uid.suffix' options to their Flink job?
6. huyuanfeng2018 commented on Dec 11, 2024

   

   huyuanfeng2018

   on Dec 11, 2024

   Contributor

   More actions

   I think #4424 can solve this problem. But do we still need to add the disable chain? Or do we just need to recommend that users add the 'sink.operator-uid.suffix' and 'source.operator-uid.suffix' options to their Flink job?

   @Tan-JiaLiang
   I think #4424 is enough to solve the problem of not being able to recover from a checkpoint, but I think #3232. is also necessary, Because if the flink adaptive scheduler is used, we can quickly modify the parallelism of the writer in flink webui, No need to restart the job(if chaining false).