wombatu-kun commented on code in PR #20156:
URL: https://github.com/apache/hudi/pull/20156#discussion_r4151263148
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java:
##########
@@ -102,129 +60,26 @@ public CompactionCommitSinkV2(Configuration conf) {
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
- if (writeClient == null) {
- this.writeClient = FlinkWriteClients.createWriteClient(conf,
getRuntimeContext());
- }
- this.commitBuffer = new HashMap<>();
- this.compactionPlanCache = new HashMap<>();
- this.table = this.writeClient.getHoodieTable();
- registerMetrics();
+ this.compactCommitHandler =
TableServiceHandlerFactory.createCompactionCommitHandler(conf,
getRuntimeContext());
+
this.compactCommitHandler.registerMetrics(getRuntimeContext().getMetricGroup());
}
@Override
public void processElement(
CompactionCommitEvent event,
ProcessFunction<CompactionCommitEvent, RowData>.Context context,
Collector<RowData> collector) throws Exception {
- final String instant = event.getInstant();
- if (event.isFailed()
- || (event.getWriteStatuses() != null
- && event.getWriteStatuses().stream().anyMatch(writeStatus ->
writeStatus.getTotalErrorRecords() > 0))) {
- log.warn("Received abnormal CompactionCommitEvent of instant {}, task ID
is {},"
- + " is failed: {}, error record count: {}",
- instant, event.getTaskID(), event.isFailed(),
getNumErrorRecords(event));
- }
- commitBuffer.computeIfAbsent(instant, k -> new HashMap<>())
- .put(event.getFileId(), event);
- commitIfNecessary(instant, commitBuffer.get(instant).values());
- }
-
- private long getNumErrorRecords(CompactionCommitEvent event) {
- if (event.getWriteStatuses() == null) {
- return -1L;
- }
- return event.getWriteStatuses().stream()
- .map(WriteStatus::getTotalErrorRecords).reduce(Long::sum).orElse(0L);
+ compactCommitHandler.commitIfNecessary(event);
}
- /**
- * Condition to commit: the commit buffer has equal size with the compaction
plan operations
- * and all the compact commit event {@link CompactionCommitEvent} has the
same compaction instant time.
- *
- * @param instant Compaction commit instant time
- * @param events Commit events ever received for the instant
- */
- private void commitIfNecessary(String instant,
Collection<CompactionCommitEvent> events) throws IOException {
- HoodieCompactionPlan compactionPlan =
compactionPlanCache.computeIfAbsent(instant, k -> {
- try {
- return CompactionUtils.getCompactionPlan(
- this.writeClient.getHoodieTable().getMetaClient(), instant);
- } catch (Exception e) {
- throw new HoodieException(e);
- }
- });
-
- boolean isReady = compactionPlan.getOperations().size() == events.size();
- if (!isReady) {
- return;
- }
-
- if (events.stream().anyMatch(CompactionCommitEvent::isFailed)) {
- try {
- // handle failure case
- CompactionUtil.rollbackCompaction(table, instant,
writeClient.getTransactionManager());
- } finally {
- // remove commitBuffer to avoid obsolete metadata commit
- reset(instant);
- this.compactionMetrics.markCompactionRolledBack();
- }
- return;
- }
-
+ @Override
+ public void close() throws Exception {
try {
- doCommit(instant, events);
- } catch (Throwable throwable) {
- // make it fail-safe
- log.error("Error while committing compaction instant: {}", instant,
throwable);
- this.compactionMetrics.markCompactionRolledBack();
+ if (compactCommitHandler != null) {
+ compactCommitHandler.close();
+ }
} finally {
- // reset the status
- reset(instant);
+ super.close();
Review Comment:
+1
--
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]
