yuzhaojing commented on code in PR #6732:
URL: https://github.com/apache/hudi/pull/6732#discussion_r1062128020
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java:
##########
@@ -352,182 +303,23 @@ public void commitLogCompaction(String
logCompactionInstantTime, HoodieCommitMet
protected void completeLogCompaction(HoodieCommitMetadata metadata,
HoodieTable table,
String logCompactionCommitTime) {
- this.context.setJobStatus(this.getClass().getSimpleName(), "Collect log
compaction write status and commit compaction");
- List<HoodieWriteStat> writeStats = metadata.getWriteStats();
final HoodieInstant logCompactionInstant = new
HoodieInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.LOG_COMPACTION_ACTION, logCompactionCommitTime);
- try {
- this.txnManager.beginTransaction(Option.of(logCompactionInstant),
Option.empty());
- preCommit(logCompactionInstant, metadata);
- finalizeWrite(table, logCompactionCommitTime, writeStats);
- // commit to data table after committing to metadata table.
- updateTableMetadata(table, metadata, logCompactionInstant);
- LOG.info("Committing Log Compaction " + logCompactionCommitTime + ".
Finished with result " + metadata);
- CompactHelpers.getInstance().completeInflightLogCompaction(table,
logCompactionCommitTime, metadata);
- } finally {
- this.txnManager.endTransaction(Option.of(logCompactionInstant));
- }
- WriteMarkersFactory.get(config.getMarkersType(), table,
logCompactionCommitTime)
- .quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
- if (compactionTimer != null) {
- long durationInMs = metrics.getDurationInMs(compactionTimer.stop());
-
HoodieActiveTimeline.parseDateFromInstantTimeSafely(logCompactionCommitTime).ifPresent(parsedInstant
->
- metrics.updateCommitMetrics(parsedInstant.getTime(), durationInMs,
metadata, HoodieActiveTimeline.LOG_COMPACTION_ACTION)
- );
- }
- LOG.info("Log Compacted successfully on commit " +
logCompactionCommitTime);
+ preCommit(logCompactionInstant, metadata);
+ tableServiceClient.completeLogCompaction(metadata, table,
logCompactionCommitTime);
Review Comment:
This causes an instant unexpected problem with concurrency, which I need to
fix.
--
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]
