danny0405 commented on code in PR #6025:
URL: https://github.com/apache/hudi/pull/6025#discussion_r912431748
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanSourceFunction.java:
##########
@@ -48,47 +60,154 @@
* as the instant time.</li>
* </ul>
*/
-public class CompactionPlanSourceFunction extends AbstractRichFunction
implements SourceFunction<CompactionPlanEvent> {
+public class CompactionPlanSourceFunction
+ extends AbstractRichFunction implements
SourceFunction<CompactionPlanEvent> {
+ private static final Logger LOG =
LoggerFactory.getLogger(CompactionPlanSourceFunction.class);
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * The interval between consecutive path scans.
+ */
+ private final long interval;
- protected static final Logger LOG =
LoggerFactory.getLogger(CompactionPlanSourceFunction.class);
+ private volatile boolean isRunning = true;
+
+ private final Configuration conf;
+
+ protected HoodieFlinkWriteClient writeClient;
/**
- * Compaction instant time.
+ * The hoodie table.
*/
- private final String compactionInstantTime;
+ private transient HoodieFlinkTable<?> table;
/**
- * The compaction plan.
+ * The path to monitor.
*/
- private final HoodieCompactionPlan compactionPlan;
+ private final transient Path path;
+
+ private final Boolean isStreamingMode;
- public CompactionPlanSourceFunction(HoodieCompactionPlan compactionPlan,
String compactionInstantTime) {
- this.compactionPlan = compactionPlan;
- this.compactionInstantTime = compactionInstantTime;
+ public CompactionPlanSourceFunction(
+ Configuration conf,
+ String path,
+ Boolean isStreamingMode) {
+ this.conf = conf;
+ this.path = new Path(path);
+ this.isStreamingMode = isStreamingMode;
+ this.interval =
conf.getInteger(FlinkOptions.COMPACTION_STREAMING_CHECK_INTERVAL);
}
@Override
public void open(Configuration parameters) throws Exception {
- // no operation
+ super.open(parameters);
+ if (writeClient == null) {
+ this.writeClient = StreamerUtil.createWriteClient(conf,
getRuntimeContext());
+ }
+ this.table = this.writeClient.getHoodieTable();
}
@Override
- public void run(SourceContext sourceContext) throws Exception {
- List<CompactionOperation> operations =
this.compactionPlan.getOperations().stream()
+ public void run(SourceContext<CompactionPlanEvent> context) throws Exception
{
+ if (isStreamingMode) {
+ while (isRunning) {
+ monitorCompactionPlan(context);
+ TimeUnit.SECONDS.sleep(interval);
+ }
+ } else {
+ monitorCompactionPlan(context);
+ }
+ }
+
+ public void monitorCompactionPlan(SourceContext<CompactionPlanEvent>
context) throws IOException {
+ table.getMetaClient().reloadActiveTimeline();
+
+ // checks the compaction plan and do compaction.
+ if (OptionsResolver.needsScheduleCompaction(conf)) {
+ Option<String> compactionInstantTimeOption =
CompactionUtil.getCompactionInstantTime(table.getMetaClient());
+ if (compactionInstantTimeOption.isPresent()) {
+ boolean scheduled =
writeClient.scheduleCompactionAtInstant(compactionInstantTimeOption.get(),
Option.empty());
+ if (!scheduled) {
+ // do nothing.
+ LOG.info("No compaction plan for this job ");
+ return;
+ }
+ table.getMetaClient().reloadActiveTimeline();
+ }
+ }
+
+ // fetch the instant based on the configured execution sequence
+ String compactionSeq = conf.getString(FlinkOptions.COMPACTION_SEQUENCE);
+ HoodieTimeline timeline =
table.getActiveTimeline().filterPendingCompactionTimeline();
+ Option<HoodieInstant> requested = CompactionUtil.isLIFO(compactionSeq) ?
timeline.lastInstant() : timeline.firstInstant();
+ if (!requested.isPresent()) {
+ // do nothing.
+ LOG.info("No compaction plan scheduled, turns on the compaction plan
schedule with --schedule option");
+ return;
+ }
+
+ String compactionInstantTime = requested.get().getTimestamp();
+
+ HoodieInstant inflightInstant =
HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
+ if (timeline.containsInstant(inflightInstant)) {
+ LOG.info("Rollback inflight compaction instant: [" +
compactionInstantTime + "]");
+ table.rollbackInflightCompaction(inflightInstant);
+ table.getMetaClient().reloadActiveTimeline();
+ }
+
+ // generate compaction plan
+ // should support configurable commit metadata
+ HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
+ table.getMetaClient(), compactionInstantTime);
+
+ if (compactionPlan == null || (compactionPlan.getOperations() == null)
+ || (compactionPlan.getOperations().isEmpty())) {
+ // No compaction plan, do nothing and return.
+ LOG.info("No compaction plan for instant " + compactionInstantTime);
+ return;
+ }
+
+ HoodieInstant instant =
HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
+ HoodieTimeline pendingCompactionTimeline =
table.getActiveTimeline().filterPendingCompactionTimeline();
+ if (!pendingCompactionTimeline.containsInstant(instant)) {
Review Comment:
Is it necessary to start a separate scheduling for compaction ? There is
only possible compaction plans when an instant is committed successfully, seems
more reasonable to schedule the compaction on ckp success event, based on the
fact users may have custom ckp interval config.
--
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]
