This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 0b4a196f32f [HUDI-7172] Fix the timeline archiver to support
concurrent writer (#10244)
0b4a196f32f is described below
commit 0b4a196f32fbc6bca1980ec693e5c929b19e54be
Author: Danny Chan <[email protected]>
AuthorDate: Wed Dec 6 07:08:20 2023 +0800
[HUDI-7172] Fix the timeline archiver to support concurrent writer (#10244)
---
.../client/timeline/ActiveActionWithDetails.java | 15 ++---
.../client/timeline/HoodieTimelineArchiver.java | 16 +++--
.../java/org/apache/hudi/DummyActiveAction.java | 4 +-
.../hudi/common/table/timeline/ActiveAction.java | 69 ++++++++++++++--------
.../table/timeline/MetadataConversionUtils.java | 6 +-
5 files changed, 69 insertions(+), 41 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/ActiveActionWithDetails.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/ActiveActionWithDetails.java
index 1065a03fa0f..33725baa4aa 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/ActiveActionWithDetails.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/ActiveActionWithDetails.java
@@ -27,6 +27,7 @@ import org.apache.hudi.common.util.collection.Pair;
import javax.annotation.Nullable;
+import java.util.Collections;
import java.util.List;
import java.util.Objects;
@@ -50,7 +51,7 @@ public class ActiveActionWithDetails extends ActiveAction {
Option<byte[]> inflightDetails,
HoodieInstant completed,
Option<byte[]> completedDetails) {
- super(requested, inflight, completed);
+ super(requested, inflight, Collections.singletonList(completed));
this.requestedDetails = requestedDetails;
this.inflightDetails = inflightDetails;
this.completedDetails = completedDetails;
@@ -96,18 +97,18 @@ public class ActiveActionWithDetails extends ActiveAction {
return this.inflightDetails;
}
- public byte[] getCleanPlan(HoodieTableMetaClient metaClient) {
+ public Option<byte[]> getCleanPlan(HoodieTableMetaClient metaClient) {
ValidationUtils.checkState(this.requestedDetails.isPresent(), "clean plan
does not exist");
- return this.requestedDetails.get();
+ return this.requestedDetails;
}
- public byte[] getCompactionPlan(HoodieTableMetaClient metaClient) {
+ public Option<byte[]> getCompactionPlan(HoodieTableMetaClient metaClient) {
ValidationUtils.checkState(this.requestedDetails.isPresent(), "compaction
plan does not exist");
- return this.requestedDetails.get();
+ return this.requestedDetails;
}
- public byte[] getLogCompactionPlan(HoodieTableMetaClient metaClient) {
+ public Option<byte[]> getLogCompactionPlan(HoodieTableMetaClient metaClient)
{
ValidationUtils.checkState(this.requestedDetails.isPresent(), "log
compaction plan does not exist");
- return this.requestedDetails.get();
+ return this.requestedDetails;
}
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/HoodieTimelineArchiver.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/HoodieTimelineArchiver.java
index cb0b748dd5d..c6bb7b300f4 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/HoodieTimelineArchiver.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/HoodieTimelineArchiver.java
@@ -295,15 +295,23 @@ public class HoodieTimelineArchiver<T extends
HoodieAvroPayload, I, K, O> {
}
// For archive, we need to include instant's all states.
+ // The re-instantiation of the timeline may result in inconsistencies with
the existing meta client active timeline,
+ // When there is no lock guard of the archiving process, the 'raw'
timeline could contain less distinct instants
+ // because of the metadata file removing from another archiving process.
HoodieActiveTimeline rawActiveTimeline = new
HoodieActiveTimeline(metaClient, false);
Map<Pair<String, String>, List<HoodieInstant>> groupByTsAction =
rawActiveTimeline.getInstantsAsStream()
.collect(Collectors.groupingBy(i -> Pair.of(i.getTimestamp(),
HoodieInstant.getComparableAction(i.getAction()))));
- return instantsToArchive.stream().map(hoodieInstant -> {
+ return instantsToArchive.stream().flatMap(hoodieInstant -> {
List<HoodieInstant> instantsToStream =
groupByTsAction.get(Pair.of(hoodieInstant.getTimestamp(),
HoodieInstant.getComparableAction(hoodieInstant.getAction())));
- return ActiveAction.fromInstants(instantsToStream);
+ if (instantsToStream != null) {
+ return Stream.of(ActiveAction.fromInstants(instantsToStream));
+ } else {
+ // if a concurrent writer archived the instant
+ return Stream.empty();
+ }
});
}
@@ -314,7 +322,7 @@ public class HoodieTimelineArchiver<T extends
HoodieAvroPayload, I, K, O> {
List<HoodieInstant> completedInstants = new ArrayList<>();
for (ActiveAction activeAction : activeActions) {
- completedInstants.add(activeAction.getCompleted());
+ completedInstants.addAll(activeAction.getCompletedInstants());
pendingInstants.addAll(activeAction.getPendingInstants());
}
@@ -347,7 +355,7 @@ public class HoodieTimelineArchiver<T extends
HoodieAvroPayload, I, K, O> {
private void deleteAnyLeftOverMarkers(HoodieEngineContext context,
ActiveAction activeAction) {
WriteMarkers writeMarkers =
WriteMarkersFactory.get(config.getMarkersType(), table,
activeAction.getInstantTime());
if (writeMarkers.deleteMarkerDir(context,
config.getMarkersDeleteParallelism())) {
- LOG.info("Cleaned up left over marker directory for instant :" +
activeAction.getCompleted());
+ LOG.info("Cleaned up left over marker directory for instant :" +
activeAction);
}
}
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/DummyActiveAction.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/DummyActiveAction.java
index 56adbac7580..0e9ca44bdd6 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/DummyActiveAction.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/DummyActiveAction.java
@@ -23,6 +23,8 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.util.Option;
+import java.util.Collections;
+
/**
* Instant triple for testing.
*/
@@ -35,7 +37,7 @@ public class DummyActiveAction extends ActiveAction {
public DummyActiveAction(HoodieInstant completed, byte[] commitMetadata) {
super(new HoodieInstant(HoodieInstant.State.REQUESTED,
completed.getAction(), completed.getTimestamp()),
new HoodieInstant(HoodieInstant.State.INFLIGHT, completed.getAction(),
completed.getTimestamp()),
- completed);
+ Collections.singletonList(completed));
this.commitMetadata = commitMetadata;
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/ActiveAction.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/ActiveAction.java
index 15f6b2de376..fe07a7ff6b0 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/ActiveAction.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/ActiveAction.java
@@ -27,8 +27,8 @@ import javax.annotation.Nullable;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.List;
-import java.util.Objects;
/**
* A combination of instants covering action states: requested, inflight,
completed.
@@ -36,33 +36,36 @@ import java.util.Objects;
public class ActiveAction implements Serializable, Comparable<ActiveAction> {
private final HoodieInstant requested;
private final HoodieInstant inflight;
- private final HoodieInstant completed;
+ private final List<HoodieInstant> completed;
/**
* The constructor.
*/
- protected ActiveAction(@Nullable HoodieInstant requested, @Nullable
HoodieInstant inflight, HoodieInstant completed) {
+ protected ActiveAction(@Nullable HoodieInstant requested, @Nullable
HoodieInstant inflight, List<HoodieInstant> completed) {
this.requested = requested;
this.inflight = inflight;
this.completed = completed;
}
public static ActiveAction fromInstants(List<HoodieInstant> instants) {
- ValidationUtils.checkArgument(instants.size() <= 3,
- "Number of instant metadata files should be <= 3: " + instants);
+ ValidationUtils.checkArgument(instants != null, "Instants should not be
null");
HoodieInstant requested = null;
HoodieInstant inflight = null;
- HoodieInstant completed = null;
+ // there could be multiple completed cleaning instants for one instant
time,
+ // currently we do not force explicit lock guard for cleaning.
+ List<HoodieInstant> completed = new ArrayList<>();
for (HoodieInstant instant : instants) {
if (instant.isRequested()) {
requested = instant;
} else if (instant.isInflight()) {
inflight = instant;
} else {
- completed = instant;
+ completed.add(instant);
}
}
- return new ActiveAction(requested, inflight,
Objects.requireNonNull(completed));
+ ValidationUtils.checkState(!completed.isEmpty(), "The instants to archive
must be completed: " + instants);
+
completed.sort(Comparator.comparing(HoodieInstant::getCompletionTime).reversed());
+ return new ActiveAction(requested, inflight, completed);
}
public List<HoodieInstant> getPendingInstants() {
@@ -76,31 +79,35 @@ public class ActiveAction implements Serializable,
Comparable<ActiveAction> {
return instants;
}
- public HoodieInstant getCompleted() {
+ public List<HoodieInstant> getCompletedInstants() {
return completed;
}
+ private HoodieInstant getCompleted() {
+ return completed.get(0);
+ }
+
public String getAction() {
- return this.completed.getAction();
+ return getCompleted().getAction();
}
/**
* A COMPACTION action eventually becomes COMMIT when completed.
*/
public String getPendingAction() {
- return getPendingInstant().getAction();
+ return getPendingInstant().map(HoodieInstant::getAction).orElse("null");
}
public String getInstantTime() {
- return this.completed.getTimestamp();
+ return getCompleted().getTimestamp();
}
public String getCompletionTime() {
- return this.completed.getCompletionTime();
+ return getCompleted().getCompletionTime();
}
public Option<byte[]> getCommitMetadata(HoodieTableMetaClient metaClient) {
- Option<byte[]> content =
metaClient.getActiveTimeline().getInstantDetails(this.completed);
+ Option<byte[]> content =
metaClient.getActiveTimeline().getInstantDetails(getCompleted());
if (content.isPresent() && content.get().length == 0) {
return Option.empty();
}
@@ -133,35 +140,45 @@ public class ActiveAction implements Serializable,
Comparable<ActiveAction> {
}
}
- public byte[] getCleanPlan(HoodieTableMetaClient metaClient) {
- return
metaClient.getActiveTimeline().readCleanerInfoAsBytes(getPendingInstant()).get();
+ public Option<byte[]> getCleanPlan(HoodieTableMetaClient metaClient) {
+ Option<HoodieInstant> pendingInstant = getPendingInstant();
+ if (!pendingInstant.isPresent()) {
+ return Option.empty();
+ }
+ return
metaClient.getActiveTimeline().readCleanerInfoAsBytes(pendingInstant.get());
}
- public byte[] getCompactionPlan(HoodieTableMetaClient metaClient) {
- return
metaClient.getActiveTimeline().readCompactionPlanAsBytes(HoodieTimeline.getCompactionRequestedInstant(getInstantTime())).get();
+ public Option<byte[]> getCompactionPlan(HoodieTableMetaClient metaClient) {
+ if (this.requested != null) {
+ return
metaClient.getActiveTimeline().readCompactionPlanAsBytes(this.requested);
+ }
+ return Option.empty();
}
- public byte[] getLogCompactionPlan(HoodieTableMetaClient metaClient) {
- return
metaClient.getActiveTimeline().readCompactionPlanAsBytes(HoodieTimeline.getLogCompactionRequestedInstant(getInstantTime())).get();
+ public Option<byte[]> getLogCompactionPlan(HoodieTableMetaClient metaClient)
{
+ if (this.requested != null) {
+ return
metaClient.getActiveTimeline().readCompactionPlanAsBytes(this.requested);
+ }
+ return Option.empty();
}
- protected HoodieInstant getPendingInstant() {
+ protected Option<HoodieInstant> getPendingInstant() {
if (requested != null) {
- return requested;
+ return Option.of(requested);
} else if (inflight != null) {
- return inflight;
+ return Option.of(inflight);
} else {
- throw new AssertionError("Pending instant does not exist.");
+ return Option.empty();
}
}
@Override
public int compareTo(ActiveAction other) {
- return
this.completed.getTimestamp().compareTo(other.completed.getTimestamp());
+ return
this.getCompleted().getTimestamp().compareTo(other.getCompleted().getTimestamp());
}
@Override
public String toString() {
- return getCompleted().getTimestamp() + "__" + getCompleted().getAction();
+ return getInstantTime() + "__" + getAction();
}
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java
index 8a3f0e0692b..fb7710eee5c 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java
@@ -154,7 +154,7 @@ public class MetadataConversionUtils {
lsmTimelineInstant.setVersion(LSMTimeline.LSM_TIMELINE_INSTANT_VERSION_1);
switch (activeAction.getPendingAction()) {
case HoodieTimeline.CLEAN_ACTION: {
-
lsmTimelineInstant.setPlan(ByteBuffer.wrap(activeAction.getCleanPlan(metaClient)));
+ activeAction.getCleanPlan(metaClient).ifPresent(plan ->
lsmTimelineInstant.setPlan(ByteBuffer.wrap(plan)));
break;
}
case HoodieTimeline.REPLACE_COMMIT_ACTION: {
@@ -168,11 +168,11 @@ public class MetadataConversionUtils {
break;
}
case HoodieTimeline.COMPACTION_ACTION: {
-
lsmTimelineInstant.setPlan(ByteBuffer.wrap(activeAction.getCompactionPlan(metaClient)));
+ activeAction.getCompactionPlan(metaClient).ifPresent(plan ->
lsmTimelineInstant.setPlan(ByteBuffer.wrap(plan)));
break;
}
case HoodieTimeline.LOG_COMPACTION_ACTION: {
-
lsmTimelineInstant.setPlan(ByteBuffer.wrap(activeAction.getLogCompactionPlan(metaClient)));
+ activeAction.getLogCompactionPlan(metaClient).ifPresent(plan ->
lsmTimelineInstant.setPlan(ByteBuffer.wrap(plan)));
break;
}
default:
