Added fail-fast option to CopyableDatasetRequestor (disabled by default)
@@ -40,6 +40,7 @@ import gobblin.data.management.copy.prioritization.PrioritizedCopyableDataset;
import gobblin.util.request_allocation.PushDownRequestor;
import javax.annotation.Nullable;
+
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -52,6 +53,8 @@ import lombok.extern.slf4j.Slf4j;
@AllArgsConstructor
@Getter
public class CopyableDatasetRequestor implements PushDownRequestor<FileSet<CopyEntity>> {
+ private final static String CONF_PREFIX = CopyConfiguration.COPY_PREFIX + "." + CopyableDatasetRequestor.class.getSimpleName();
+ private final static String FAIL_FAST = CONF_PREFIX + ".failFast";
@AllArgsConstructor
public static class Factory implements Function<CopyableDatasetBase, CopyableDatasetRequestor> {
@@ -86,6 +89,10 @@ public class CopyableDatasetRequestor implements PushDownRequestor<FileSet<CopyE
try {
return injectRequestor(this.dataset.getFileSetIterator(this.targetFs, this.copyConfiguration));
} catch (Throwable exc) {
+ if (copyConfiguration.getConfig().hasPath(FAIL_FAST)
+ && copyConfiguration.getConfig().getBoolean(FAIL_FAST)) {
+ throw new RuntimeException(String.format("Could not get FileSets for dataset %s", this.dataset.datasetURN()), exc);
+ }
log.error(String.format("Could not get FileSets for dataset %s. Skipping.", this.dataset.datasetURN()), exc);
return Iterators.emptyIterator();
}
