lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-3257] Excluding clustering instants from pending rollback info (#4616)

Browse Source
This commit is contained in:
Danny Chan
2022-01-17 18:18:45 +08:00
committed by GitHub
parent 75caa7d3d8
commit 36a9f63e45
1 changed files with 8 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
View File

@@ -906,14 +906,18 @@ public abstract class AbstractHoodieWriteClient<T extends HoodieRecordPayload, I
protected Map<String, Option<HoodiePendingRollbackInfo>> getPendingRollbackInfos(HoodieTableMetaClient metaClient) { protected Map<String, Option<HoodiePendingRollbackInfo>> getPendingRollbackInfos(HoodieTableMetaClient metaClient) {
List<HoodieInstant> instants = metaClient.getActiveTimeline().filterPendingRollbackTimeline().getInstants().collect(Collectors.toList()); List<HoodieInstant> instants = metaClient.getActiveTimeline().filterPendingRollbackTimeline().getInstants().collect(Collectors.toList());
Map<String, Option<HoodiePendingRollbackInfo>> infoMap = new HashMap<>(); Map<String, Option<HoodiePendingRollbackInfo>> infoMap = new HashMap<>();
HoodieTimeline pendingCompactionTimeline = metaClient.getActiveTimeline().filterPendingCompactionTimeline();
for (HoodieInstant instant : instants) { for (HoodieInstant instant : instants) {
try { try {
HoodieRollbackPlan rollbackPlan = RollbackUtils.getRollbackPlan(metaClient, instant); HoodieRollbackPlan rollbackPlan = RollbackUtils.getRollbackPlan(metaClient, instant);
String action = rollbackPlan.getInstantToRollback().getAction();
if (!HoodieTimeline.COMPACTION_ACTION.equals(action)) {
boolean isClustering = HoodieTimeline.REPLACE_COMMIT_ACTION.equals(action)
&& ClusteringUtils.getClusteringPlan(metaClient, instant).isPresent();
if (!isClustering) {
String instantToRollback = rollbackPlan.getInstantToRollback().getCommitTime(); String instantToRollback = rollbackPlan.getInstantToRollback().getCommitTime();
if (!pendingCompactionTimeline.containsInstant(instantToRollback)) {
infoMap.putIfAbsent(instantToRollback, Option.of(new HoodiePendingRollbackInfo(instant, rollbackPlan))); infoMap.putIfAbsent(instantToRollback, Option.of(new HoodiePendingRollbackInfo(instant, rollbackPlan)));
} }
}
} catch (IOException e) { } catch (IOException e) {
LOG.warn("Fetching rollback plan failed for " + infoMap + ", skip the plan", e); LOG.warn("Fetching rollback plan failed for " + infoMap + ", skip the plan", e);
} }