cshuo commented on code in PR #20156:
URL: https://github.com/apache/hudi/pull/20156#discussion_r4143414026
##########
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:
If compactCommitHandler.close() throws and super.close() also throws, the
exception from this new finally block replaces the original commit-handler
exception without preserving it as suppressed. Both compaction adapters
introduce this behavior. Please use a suppression-preserving close operation,
such as IOUtils.closeAll() over the two close actions, so cleanup still
attempts both resources and retains the original failure.
--
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]
