前言
Hudi Flink支持配置table service的异步执行。Schedule的时机为checkpoint完成的时候。执行过程在线程池中完成。Flink Hudi 常用的table service有compaction，clustering和clean三种。它们对应的配置项为：
- clustering.async.enabled：是否开启异步的clustering。默认不开启。
- compaction.async.enabled：是否开启异步compaction。默认开启。
- clean.async.enabled：是否开启异步clean。默认开启。
本篇主要分析Flink中Hudi table service的排期和执行时机。至于compaction clustering和clean表服务具体的执行逻辑，参见：
- Hudi Compaction使用和源码分析 - 简书 (jianshu.com)
- Hudi 源码之 Clustering - 简书 (jianshu.com)
- Hudi 源码之 Cleaning service - 简书 (jianshu.com)
Scheduling排期
Hudi Flink table service的排期主要位于如下两个方法中：
- StreamWriteOperatorCoordinator::handleInputEvent： batch模式需要schedule的table service。
- StreamWriteOperatorCoordinator::notifyCheckpointComplete： streaming模式需要schedule的table service
接下来我们分别分析这两个方法。
handleEndInputWvent方法：
private void handleEndInputEvent(WriteMetadataEvent event) {
addEventToBuffer(event);
// 如果已经接收到所有数据
if (allEventsReceived()) {
// start to commit the instant.
// 提交
// 如果数据成功写入，返回true，表示提交成功
boolean committed = commitInstant(this.instant);
if (committed) {
// The executor thread inherits the classloader of the #handleEventFromOperator
// caller, which is a AppClassLoader. Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
// sync Hive synchronously if it is enabled in batch mode.
// 如果开启了hive sync，执行
syncHive();
// schedules the compaction or clustering if it is enabled in batch execution mode
// 表服务排期
scheduleTableServices(true);
}
}
}
notifyCheckpointComplete方法。在checkpoint执行成功的时候执行回调。
@Override
public void notifyCheckpointComplete(long checkpointId) {
executor.execute(
() -> {
// The executor thread inherits the classloader of the #notifyCheckpointComplete
// caller, which is a AppClassLoader.
Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
// for streaming mode, commits the ever received events anyway,
// the stream write task snapshot and flush the data buffer synchronously in sequence,
// so a successful checkpoint subsumes the old one(follows the checkpoint subsuming contract)
// 如果数据成功写入，返回true，表示提交成功
final boolean committed = commitInstant(this.instant, checkpointId);
// schedules the compaction or clustering if it is enabled in stream execution mode
// 排期表服务
scheduleTableServices(committed);
if (committed) {
// start new instant.
// 写入instant
startInstant();
// sync Hive if is enabled
// 如果开启了hive sync，执行
syncHiveAsync();
}
}, "commits the instant %s", this.instant
);
}
scheduleTableServices方法：
private void scheduleTableServices(Boolean committed) {
// if compaction is on, schedule the compaction
// 如果是MOR表，并且开启了compaction.schedule.enabled配置（默认开启）
if (tableState.scheduleCompaction) {
CompactionUtil.scheduleCompaction(writeClient, tableState.isDeltaTimeCompaction, committed);
}
// if clustering is on, schedule the clustering
// clustering.schedule.enabled如果开启（默认不开启）
// 对于bucket index表，如果配置的是consistent hash（一致性hash），要求写入类型必须是upsert
// 否则（SIMPLE类型）要求写入类型必须是insert
if (tableState.scheduleClustering) {
ClusteringUtil.scheduleClustering(conf, writeClient, committed);
}
}
Executing执行
HoodieTableSink
Hudi Flink创建table service异步任务流位于HoodieTableSink::getSinkRuntimeProvider。
// ...
// Append mode
// 如果是增量写入模式
if (OptionsResolver.isAppendMode(conf)) {
// close compaction for append mode
// 关闭compaction schedule模式
conf.set(FlinkOptions.COMPACTION_SCHEDULE_ENABLED, false);
// append 模式写入数据
DataStream<Object> pipeline = Pipelines.append(conf, rowType, dataStream);
// 如果需要异步clustering
// write.operation为insert并且启用了异步clustering（clustering.async.enabled为true）
if (OptionsResolver.needsAsyncClustering(conf)) {
// 执行clustering
return Pipelines.cluster(conf, rowType, pipeline);
// 如果hoodie.cleaner.policy.failed.writes配置为lazy
} else if (OptionsResolver.isLazyFailedWritesCleanPolicy(conf)) {
// add clean function to rollback failed writes for lazy failed writes cleaning policy
// 执行清理
return Pipelines.clean(conf, pipeline);
} else {
// 否则什么也不做
return Pipelines.dummySink(pipeline);
}
}
DataStream<Object> pipeline;
// bootstrap加载索引
final DataStream<HoodieRecord> hoodieRecordDataStream =
Pipelines.bootstrap(conf, rowType, dataStream, context.isBounded(), overwrite);
// write pipeline
// 流式写入
pipeline = Pipelines.hoodieStreamWrite(conf, hoodieRecordDataStream);
// compaction
// 是否需要异步压缩
// compaction.async.enabled是否为true。默认为true
if (OptionsResolver.needsAsyncCompaction(conf)) {
// use synchronous compaction for bounded source.
// 如果是bounded数据源（有头有尾），使用同步压缩
if (context.isBounded()) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
// 执行压缩
return Pipelines.compact(conf, pipeline);
} else {
// 如果没有配置压缩，执行清理
return Pipelines.clean(conf, pipeline);
}
// ...
通过上面的分析不难得知，Flink中compact，clean和clustering表服务都在Pipeline中创建。接下来我们分析Pipeline的源代码。
Pipelines
Pipeline创建了一条专用的数据流，这些数据流分别用来周期性创建compaction和clustering的执行计划，以及执行compact，clean和clustering。它们独立于系统的业务数据流。
Pipelines::compact
该方法用来启动周期压缩任务流。
public static DataStreamSink<CompactionCommitEvent> compact(Configuration conf, DataStream<Object> dataStream) {
// 使用CompactionPlanOperator下发compaction执行计划
DataStreamSink<CompactionCommitEvent> compactionCommitEventDataStream = dataStream.transform("compact_plan_generate",
TypeInformation.of(CompactionPlanEvent.class),
new CompactionPlanOperator(conf))
// plan生成过程必须是单并行度
.setParallelism(1) // plan generate must be singleton
.setMaxParallelism(1)
// make the distribution strategy deterministic to avoid concurrent modifications
// on the same bucket files .keyBy(plan -> plan.getOperation().getFileGroupId().getFileId())
// 使用CompactorOperator执行压缩计划
.transform("compact_task",
TypeInformation.of(CompactionCommitEvent.class),
new CompactOperator(conf))
// 并行度配置为compaction.tasks
.setParallelism(conf.getInteger(FlinkOptions.COMPACTION_TASKS))
// CompactionSommitSink检查并提交compaction instant
.addSink(new CompactionCommitSink(conf))
.name("compact_commit")
// 执行commit的并行度必须是1
.setParallelism(1); // compaction commit should be singleton
compactionCommitEventDataStream.getTransformation().setMaxParallelism(1);
return compactionCommitEventDataStream;
}
CompactionPlanOperator::notifyCheckpointComplete。在checkpoint的时候检查是否生成的有requested状态的compaction instant。如果有，生成CompactionPlanEvent发往下游。
SteamWriteOperatorCoordinator用来生成requested状态的compaction instant，CompactionPlanOperator用来获取到这些compaction instant，读取保存的执行计划然后发往下游。
@Override
public void notifyCheckpointComplete(long checkpointId) {
try {
table.getMetaClient().reloadActiveTimeline();
// There is no good way to infer when the compaction task for an instant crushed
// or is still undergoing. So we use a configured timeout threshold to control the rollback: // {@code FlinkOptions.COMPACTION_TIMEOUT_SECONDS}, // when the earliest inflight instant has timed out, assumes it has failed // already and just rolls it back.
// comment out: do we really need the timeout rollback ? // CompactionUtil.rollbackEarliestCompaction(table, conf);
scheduleCompaction(table, checkpointId);
} catch (Throwable throwable) {
// make it fail-safe
LOG.error("Error while scheduling compaction plan for checkpoint: " + checkpointId, throwable);
}
}
scheduleCompaction方法读取第一个状态为reqested状态的compaction instant，获取到它的compaction plan，将该compaction涉及到的compaction plan中的compactionOperation（即Compaction操作涉及到的file group信息）包装为CompactionPlanEvent发往下游。
private void scheduleCompaction(HoodieFlinkTable<?> table, long checkpointId) throws IOException {
// 获取包含所有pending compaction的timeline
HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
// the first instant takes the highest priority.
// 找到时间最早的requested compaction instant
Option<HoodieInstant> firstRequested = pendingCompactionTimeline
.filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED).firstInstant();
// record metrics
compactionMetrics.setFirstPendingCompactionInstant(firstRequested);
compactionMetrics.setPendingCompactionCount(pendingCompactionTimeline.countInstants());
// 如果没有requested状态的compaction instant，说明没有必要schedule
if (!firstRequested.isPresent()) {
// do nothing.
LOG.info("No compaction plan for checkpoint " + checkpointId);
return;
}
// 获取这个requested compaction instant对应的时间
String compactionInstantTime = firstRequested.get().getTimestamp();
// generate compaction plan
// should support configurable commit metadata
获取compaction plan
HoodieCompactionPlan compactionPlan = CompactionUtils.getCompactionPlan(
table.getMetaClient(), compactionInstantTime);
// 如果没有获取到有效的compactio plan
if (compactionPlan == null || (compactionPlan.getOperations() == null)
|| (compactionPlan.getOperations().isEmpty())) {
// do nothing.
LOG.info("Empty compaction plan for instant " + compactionInstantTime);
} else {
// 获取这个requested状态的 compaction instant
HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
// Mark instant as compaction inflight
// 将它的状态修改为inflight
table.getActiveTimeline().transitionCompactionRequestedToInflight(instant);
table.getMetaClient().reloadActiveTimeline();
// 获取plan中所有的compaction operation，封装为compactionPlanEvent发往下游
List<CompactionOperation> operations = compactionPlan.getOperations().stream()
.map(CompactionOperation::convertFromAvroRecordInstance).collect(toList());
LOG.info("Execute compaction plan for instant {} as {} file groups", compactionInstantTime, operations.size());
// 删除标记文件
WriteMarkersFactory
.get(table.getConfig().getMarkersType(), table, compactionInstantTime)
.deleteMarkerDir(table.getContext(), table.getConfig().getMarkersDeleteParallelism());
// 每一个operation封装为一个CompactionPlanEvent
// 这样到下游的时候可以将这些压缩任务均分
// 每个并行度处理一部分file group的压缩
for (CompactionOperation operation : operations) {
output.collect(new StreamRecord<>(new CompactionPlanEvent(compactionInstantTime, operation)));
}
}
}
CompactOperator的processElement方法接收上面生成的CompactionPlanEvent，执行压缩任务。
@Override
public void processElement(StreamRecord<CompactionPlanEvent> record) throws Exception {
final CompactionPlanEvent event = record.getValue();
// 获取inflight compaction instant time
final String instantTime = event.getCompactionInstantTime();
// 获取compaction operation
final CompactionOperation compactionOperation = event.getOperation();
// 如果是异步压缩
// 在线程池中执行压缩，不会影响checkpoint过程
if (asyncCompaction) {
// executes the compaction task asynchronously to not block the checkpoint barrier propagate.
executor.execute(
() -> doCompaction(instantTime, compactionOperation, collector, reloadWriteConfig()),
(errMsg, t) -> collector.collect(new CompactionCommitEvent(instantTime, compactionOperation.getFileId(), taskID)),
"Execute compaction for instant %s from task %d", instantTime, taskID);
} else {
// executes the compaction task synchronously for batch mode.
// 否则同步执行compaction
LOG.info("Execute compaction for instant {} from task {}", instantTime, taskID);
doCompaction(instantTime, compactionOperation, collector, writeClient.getConfig());
}
}
接下来的doCompaction方法间接调用了HoodieFlinkMergeOnReadTableCompactor的compact方法，前面的文章已有分析，这里不再赘述。
Pipelines::cluster
该方法启动周期clustering任务流。
public static DataStreamSink<ClusteringCommitEvent> cluster(Configuration conf, RowType rowType, DataStream<Object> dataStream) {
// 使用ClusteringPlanOperator下发clustering执行计划
DataStream<ClusteringCommitEvent> clusteringStream = dataStream.transform("cluster_plan_generate",
TypeInformation.of(ClusteringPlanEvent.class),
new ClusteringPlanOperator(conf))
// 下发执行计划的并行度必须是1
.setParallelism(1) // plan generate must be singleton
.setMaxParallelism(1) // plan generate must be singleton
.keyBy(plan ->
// make the distribution strategy deterministic to avoid concurrent modifications
// on the same bucket files
// 按照ClusteringPlanEvent的fileId分组
// 针对同一个file slice的clustering操作会分配给相同的线程执行，防止并发修改
plan.getClusteringGroupInfo().getOperations() .stream().map(ClusteringOperation::getFileId.collect(Collectors.joining()))
.transform("clustering_task",
TypeInformation.of(ClusteringCommitEvent.class),
// 通过ClusteringOperator执行clustering
new ClusteringOperator(conf, rowType))
// clustering任务的并行度为clustering.tasks
.setParallelism(conf.getInteger(FlinkOptions.CLUSTERING_TASKS));
// 如果启用了排序，即clustering.plan.strategy.sort.columns配置项不为空
// 配置该步骤的执行内存，对应配置项为write.sort.memory
if (OptionsResolver.sortClusteringEnabled(conf)) {
ExecNodeUtil.setManagedMemoryWeight(clusteringStream.getTransformation(),
conf.getInteger(FlinkOptions.WRITE_SORT_MEMORY) * 1024L * 1024L);
}
// 检查并提交clustering instant
DataStreamSink<ClusteringCommitEvent> clusteringCommitEventDataStream = clusteringStream.addSink(new ClusteringCommitSink(conf))
.name("clustering_commit")
.setParallelism(1); // clustering commit should be singleton
clusteringCommitEventDataStream.getTransformation().setMaxParallelism(1);
return clusteringCommitEventDataStream;
}
ClusteringPlanOperatorschedule clustering的过程和前面schedule compaction的非常相似。
private void scheduleClustering(HoodieFlinkTable<?> table, long checkpointId) {
// 获取request状态的clustering instant
List<HoodieInstant> pendingClusteringInstantTimes =
ClusteringUtils.getPendingClusteringInstantTimes(table.getMetaClient());
// the first instant takes the highest priority.
// 获取时间最早的一个
Option<HoodieInstant> firstRequested = Option.fromJavaOptional(
pendingClusteringInstantTimes.stream()
.filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED).findFirst());
// record metrics
clusteringMetrics.setFirstPendingClusteringInstant(firstRequested);
clusteringMetrics.setPendingClusteringCount(pendingClusteringInstantTimes.size());
if (!firstRequested.isPresent()) {
// do nothing.
LOG.info("No clustering plan for checkpoint " + checkpointId);
return;
}
String clusteringInstantTime = firstRequested.get().getTimestamp();
// generate clustering plan
// should support configurable commit metadata HoodieInstant clusteringInstant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
// 拿到之前生成的clustering plan
Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
table.getMetaClient(), clusteringInstant);
// 如果没有获取到有效的clustering plan，直接返回
if (!clusteringPlanOption.isPresent()) {
// do nothing.
LOG.info("No clustering plan scheduled");
return;
}
HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
if (clusteringPlan == null || (clusteringPlan.getInputGroups() == null)
|| (clusteringPlan.getInputGroups().isEmpty())) {
// do nothing.
LOG.info("Empty clustering plan for instant " + clusteringInstantTime);
} else {
// Mark instant as clustering inflight
table.getActiveTimeline().transitionReplaceRequestedToInflight(clusteringInstant, Option.empty());
table.getMetaClient().reloadActiveTimeline();
// 遍历所有的inputGroup，封装为ClusteringPlanEvent
for (HoodieClusteringGroup clusteringGroup : clusteringPlan.getInputGroups()) {
LOG.info("Execute clustering plan for instant {} as {} file slices", clusteringInstantTime, clusteringGroup.getSlices().size());
output.collect(new StreamRecord<>(
new ClusteringPlanEvent(clusteringInstantTime, ClusteringGroupInfo.create(clusteringGroup), clusteringPlan.getStrategy().getStrategyParams())
));
}
}
}
Clustering对应一个file slice的操作封装为了HoodieClusteringGroup（和compatcion的CompactionOperation对应）。这里将同一个plan中所有的HoodieClusteringGroup，每一个封装为ClusteringPlanEvent，目的是为了下游可以并行执行clustering。
ClusteringOperator的processElement方法执行clustering计划。和compaction相同，分为同步执行和异步执行两种方式。
@Override
public void processElement(StreamRecord<ClusteringPlanEvent> element) throws Exception {
final ClusteringPlanEvent event = element.getValue();
final String instantTime = event.getClusteringInstantTime();
final List<ClusteringOperation> clusteringOperations = event.getClusteringGroupInfo().getOperations();
if (this.asyncClustering) {
// executes the compaction task asynchronously to not block the checkpoint barrier propagate.
executor.execute(
() -> doClustering(instantTime, clusteringOperations),
(errMsg, t) -> collector.collect(new ClusteringCommitEvent(instantTime, getFileIds(clusteringOperations), taskID)),
"Execute clustering for instant %s from task %d", instantTime, taskID);
} else {
// executes the clustering task synchronously for batch mode.
LOG.info("Execute clustering for instant {} from task {}", instantTime, taskID);
doClustering(instantTime, clusteringOperations);
}
}
doClustering方法为clustering过程的纯Flink实现。
private void doClustering(String instantTime, List<ClusteringOperation> clusteringOperations) throws Exception {
clusteringMetrics.startClustering();
// 采用bulk insert的方式写入clustering之后的数据
BulkInsertWriterHelper writerHelper = new BulkInsertWriterHelper(this.conf, this.table, this.writeConfig,
instantTime, this.taskID, getRuntimeContext().getNumberOfParallelSubtasks(), getRuntimeContext().getAttemptNumber(),
this.rowType, true);
Iterator<RowData> iterator;
// 如果clustering操作涉及到log文件，使用readRecordsForGroupWithLogs
// 否则仅使用readRecordsForGroupBaseFiles
// 这两个方法读取file group中的数据，以iterator的形式返回
if (clusteringOperations.stream().anyMatch(operation -> CollectionUtils.nonEmpty(operation.getDeltaFilePaths()))) {
// if there are log files, we read all records into memory for a file group and apply updates.
iterator = readRecordsForGroupWithLogs(clusteringOperations, instantTime);
} else {
// We want to optimize reading records for case there are no log files.
iterator = readRecordsForGroupBaseFiles(clusteringOperations);
}
// 如果配置了clustering.plan.strategy.sort.columns
// 说明需要排序
if (this.sortClusteringEnabled) {
RowDataSerializer rowDataSerializer = new RowDataSerializer(rowType);
// 使用BinaryexternalSorter来排序
// BinaryExternalSorter根据clustering.plan.strategy.sort.columns
// 生成排序代码
BinaryExternalSorter sorter = initSorter();
while (iterator.hasNext()) {
RowData rowData = iterator.next();
BinaryRowData binaryRowData = rowDataSerializer.toBinaryRow(rowData).copy();
sorter.write(binaryRowData);
}
// 使用bulk insert写入排序之后的数据
BinaryRowData row = binarySerializer.createInstance();
while ((row = sorter.getIterator().next(row)) != null) {
writerHelper.write(row);
}
sorter.close();
} else {
while (iterator.hasNext()) {
writerHelper.write(iterator.next());
}
}
List<WriteStatus> writeStatuses = writerHelper.getWriteStatuses(this.taskID);
clusteringMetrics.endClustering();
collector.collect(new ClusteringCommitEvent(instantTime, getFileIds(clusteringOperations), writeStatuses, this.taskID));
writerHelper.close();
}
Pipelines::clean
clean方法的实现较为简单。Flink为数据流增加了一个CleanFunction类型的sink，并行度为1。代码如下所示。
public static DataStreamSink<Object> clean(Configuration conf, DataStream<Object> dataStream) {
DataStreamSink<Object> cleanCommitDataStream = dataStream.addSink(new CleanFunction<>(conf))
.setParallelism(1)
.name("clean_commits");
cleanCommitDataStream.getTransformation().setMaxParallelism(1);
return cleanCommitDataStream;
我们继续分析CleanFunction。如果启用了异步clean，CleanFunction在启动的时候(open)异步执行一次。创建checkpoint的时候snapshotState启动异步clean服务。在checkpoint完成的时候notifyCheckpointComplete等待clean错操作执行完毕。相关代码如下所示。
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
this.writeClient = FlinkWriteClients.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
String instantTime = writeClient.createNewInstantTime();
LOG.info(String.format("exec clean with instant time %s...", instantTime));
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED)) {
executor.execute(() -> {
this.isCleaning = true;
try {
this.writeClient.clean(instantTime);
} finally {
this.isCleaning = false;
}
}, "wait for cleaning finish");
}
}
@Override
public void notifyCheckpointComplete(long l) throws Exception {
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && isCleaning) {
executor.execute(() -> {
try {
this.writeClient.waitForCleaningFinish();
} finally {
// ensure to switch the isCleaning flag
this.isCleaning = false;
}
}, "wait for cleaning finish");
}
}
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && !isCleaning) {
try {
this.writeClient.startAsyncCleaning();
this.isCleaning = true;
} catch (Throwable throwable) {
// catch the exception to not affect the normal checkpointing
LOG.warn("Error while start async cleaning", throwable);
}
}
}
除了Pipeline::clean方法直接使用CleanFunction之外，我们还注意到ClusteringCommitSink和CompactionCommitSink都继承了CleanFunction，都没有重写snapshotState和notifyCheckPointComplete方法。因此这两个sink的行为和CleanFunction一致，在checkpoint的时候会触发clean操作。
继续观察这两个sink的doCommit方法，我们发现最后一段都有如下代码：
// Whether to clean up the old log file when compaction
if (!conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && !isCleaning) {
this.writeClient.clean();
}
该段代码表示，如果没有启用异步clean，且当前时刻没有clean没有正在执行，执行同步clean操作。
接下来我们分析this.writeClient.startAsyncCleaning()调用，一路跟踪下去。跟踪过程中的非关键代码这里不再展示。
- CleanFunction的this.writeClient.startAsyncCleaning()
- HoodieFlinkWriteClient的tableServiceClient.startAsyncCleanerService(this);
- BaseHoodieTableServiceClient的this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(writeClient);
分析到这里，我们确定了clean表服务是在AsyncCleanerService中封装的。继续分析startAsyncCleaningIfEnabled方法，该方法首先判断是否开启了异步clean配置，如果开启了，创建一个异步clean服务，代码如下所示：
public static AsyncCleanerService startAsyncCleaningIfEnabled(BaseHoodieWriteClient writeClient) {
HoodieWriteConfig config = writeClient.getConfig();
// hoodie.clean.automatic和hoodie.clean.async这两个任意一个配置为false，不运行该服务
if (!config.isAutoClean() || !config.isAsyncClean()) {
LOG.info("The HoodieWriteClient is not configured to auto & async clean. Async clean service will not start.");
return null;
}
// 创建并启动AsyncCleanerService
AsyncCleanerService asyncCleanerService = new AsyncCleanerService(writeClient);
asyncCleanerService.start(null);
return asyncCleanerService;
}
我们继续分析asyncCleanerService.start，它位于HoodieAsyncService的start方法中。
其中的startService方法将异步服务逻辑本身和运行异步服务的executor封装为Pair返回。
public void start(Function<Boolean, Boolean> onShutdownCallback) {
if (started) {
LOG.warn("The async service already started.");
return;
}
Pair<CompletableFuture, ExecutorService> res = startService();
future = res.getKey();
executor = res.getValue();
started = true;
shutdownCallback(onShutdownCallback);
}
startService方法在实现类AsyncCleanerFunction中，如下所示。
@Override
protected Pair<CompletableFuture, ExecutorService> startService() {
String instantTime = writeClient.createNewInstantTime();
LOG.info(String.format("Starting async clean service with instant time %s...", instantTime));
//在线程池中启动异步clean操作
return Pair.of(CompletableFuture.supplyAsync(() -> {
writeClient.clean(instantTime);
return true;
}, executor), executor);
}
调用完该方法之后，异步clean的逻辑封装赋值到future变量中。
按照上面的分析notifyCheckpointComplete的时候执行waitForCompletion方法。如果clean操作还没有结束，这里阻塞等待其执行完毕。
public static void waitForCompletion(AsyncCleanerService asyncCleanerService) {
if (asyncCleanerService != null) {
LOG.info("Waiting for async clean service to finish");
try {
asyncCleanerService.waitForShutdown();
} catch (Exception e) {
throw new HoodieException("Error waiting for async clean service to finish", e);
}
}
}
分析HoodieAsnycService的waitForshutdown方法，同步等待clean执行完毕，内容如下：
public void waitForShutdown() throws ExecutionException, InterruptedException {
if (future == null) {
return;
}
try {
future.get();
} catch (ExecutionException ex) {
LOG.error("Service shutdown with error", ex);
throw ex;
}
}
参考文献
Compaction | Apache Hudi
Cleaning | Apache Hudi
Clustering | Apache Hudi
网友评论