Apache/druid
1
0
Fork 0
mirror of https://github.com/apache/druid.git synced 2025-03-01 14:59:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merger: Fix VersionConverterTask version selection by making SegmentInsertAction more flexible

Browse Source
This commit is contained in:
Gian Merlino 2013-03-14 10:37:40 -07:00
parent a1c823402b
commit 055a98a89c
2 changed files with 27 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
merger/src/main/java/com/metamx/druid/merger/common/actions/SegmentInsertAction.java
View File

@ -1,5 +1,6 @@
package com.metamx.druid.merger.common.actions; package com.metamx.druid.merger.common.actions;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.google.common.base.Predicate; import com.google.common.base.Predicate;
import com.google.common.base.Throwables; import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableSet; import com.google.common.collect.ImmutableSet;
@ -18,14 +19,25 @@ import java.util.Set;
public class SegmentInsertAction implements TaskAction<Void> public class SegmentInsertAction implements TaskAction<Void>
{ {
@JsonIgnore
private final Set<DataSegment> segments; private final Set<DataSegment> segments;
@JsonIgnore
private final boolean allowOlderVersions;
public SegmentInsertAction(Set<DataSegment> segments)
{
this(segments, false);
}
@JsonCreator @JsonCreator
public SegmentInsertAction( public SegmentInsertAction(
@JsonProperty("segments") Set<DataSegment> segments @JsonProperty("segments") Set<DataSegment> segments,
@JsonProperty("allowOlderVersions") boolean allowOlderVersions
) )
{ {
this.segments = ImmutableSet.copyOf(segments); this.segments = ImmutableSet.copyOf(segments);
this.allowOlderVersions = allowOlderVersions;
} }
@JsonProperty @JsonProperty
@ -34,6 +46,17 @@ public class SegmentInsertAction implements TaskAction<Void>
return segments; return segments;
} }
@JsonProperty
public boolean isAllowOlderVersions()
{
return allowOlderVersions;
}
public SegmentInsertAction withAllowOlderVersions(boolean _allowOlderVersions)
{
return new SegmentInsertAction(segments, _allowOlderVersions);
}
public TypeReference<Void> getReturnTypeReference() public TypeReference<Void> getReturnTypeReference()
{ {
return new TypeReference<Void>() {}; return new TypeReference<Void>() {};
@ -42,7 +65,7 @@ public class SegmentInsertAction implements TaskAction<Void>
@Override @Override
public Void perform(Task task, TaskActionToolbox toolbox) public Void perform(Task task, TaskActionToolbox toolbox)
{ {
if(!toolbox.taskLockCoversSegments(task, segments, false)) { if(!toolbox.taskLockCoversSegments(task, segments, allowOlderVersions)) {
throw new ISE("Segments not covered by locks for task: %s", task.getId()); throw new ISE("Segments not covered by locks for task: %s", task.getId());
} }

3
merger/src/main/java/com/metamx/druid/merger/common/task/VersionConverterTask.java
View File

@ -236,7 +236,8 @@ public class VersionConverterTask extends AbstractTask
DataSegment updatedSegment = segment.withVersion(String.format("%s_v%s", segment.getVersion(), outVersion)); DataSegment updatedSegment = segment.withVersion(String.format("%s_v%s", segment.getVersion(), outVersion));
updatedSegment = toolbox.getSegmentPusher().push(outLocation, updatedSegment); updatedSegment = toolbox.getSegmentPusher().push(outLocation, updatedSegment);
toolbox.getTaskActionClient().submit(new SegmentInsertAction(Sets.newHashSet(updatedSegment))); toolbox.getTaskActionClient()
.submit(new SegmentInsertAction(Sets.newHashSet(updatedSegment)).withAllowOlderVersions(true));
} else { } else {
log.info("Conversion failed."); log.info("Conversion failed.");
} }