You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@falcon.apache.org by aj...@apache.org on 2015/06/04 08:14:04 UTC
[2/2] falcon git commit: FALCON-1039 Add instance dependency API in
falcon. Contributed by Ajay Yadava
FALCON-1039 Add instance dependency API in falcon. Contributed by Ajay Yadava
Project: http://git-wip-us.apache.org/repos/asf/falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/falcon/commit/9fd86b78
Tree: http://git-wip-us.apache.org/repos/asf/falcon/tree/9fd86b78
Diff: http://git-wip-us.apache.org/repos/asf/falcon/diff/9fd86b78
Branch: refs/heads/master
Commit: 9fd86b786195ac03fc25d31c6f35062c4014f10a
Parents: 42f175a
Author: Ajay Yadava <aj...@gmail.com>
Authored: Thu Jun 4 11:42:59 2015 +0530
Committer: Ajay Yadava <aj...@gmail.com>
Committed: Thu Jun 4 11:43:32 2015 +0530
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../java/org/apache/falcon/ResponseHelper.java | 14 +
.../java/org/apache/falcon/cli/FalconCLI.java | 21 +-
.../org/apache/falcon/client/FalconClient.java | 18 +
.../org/apache/falcon/resource/EntityList.java | 6 +-
.../resource/InstanceDependencyResult.java | 86 +++++
.../resource/SchedulableEntityInstance.java | 155 ++++++++
.../org/apache/falcon/entity/EntityUtil.java | 142 ++++++-
.../org/apache/falcon/entity/FeedHelper.java | 287 +++++++++++++-
.../org/apache/falcon/entity/ProcessHelper.java | 108 ++++++
.../apache/falcon/entity/FeedHelperTest.java | 370 ++++++++++++++++++-
.../apache/falcon/entity/ProcessHelperTest.java | 207 +++++++++++
docs/src/site/twiki/FalconCLI.twiki | 35 ++
.../site/twiki/restapi/InstanceDependency.twiki | 49 +++
docs/src/site/twiki/restapi/ResourceList.twiki | 1 +
.../falcon/resource/AbstractEntityManager.java | 4 +-
.../resource/AbstractInstanceManager.java | 79 +++-
.../resource/proxy/InstanceManagerProxy.java | 37 +-
.../apache/falcon/resource/InstanceManager.java | 20 +-
.../resource/SchedulableEntityManager.java | 9 +-
.../java/org/apache/falcon/cli/FalconCLIIT.java | 4 +
21 files changed, 1624 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 9c84f85..7ce2ba3 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -4,6 +4,7 @@ Trunk (Unreleased)
INCOMPATIBLE CHANGES
NEW FEATURES
+ FALCON-1039 Add instance dependency API in falcon(Ajay Yadava)
IMPROVEMENTS
FALCON-1060 Handle transaction failures in Lineage(Pavan Kumar Kolamuri via Ajay Yadava)
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/client/src/main/java/org/apache/falcon/ResponseHelper.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/falcon/ResponseHelper.java b/client/src/main/java/org/apache/falcon/ResponseHelper.java
index 2261ceb..78598ba 100644
--- a/client/src/main/java/org/apache/falcon/ResponseHelper.java
+++ b/client/src/main/java/org/apache/falcon/ResponseHelper.java
@@ -24,6 +24,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.resource.FeedInstanceResult;
import org.apache.falcon.resource.FeedLookupResult;
+import org.apache.falcon.resource.InstanceDependencyResult;
import org.apache.falcon.resource.InstancesResult;
import org.apache.falcon.resource.InstancesSummaryResult;
import org.apache.falcon.resource.EntitySummaryResult;
@@ -277,4 +278,17 @@ public final class ResponseHelper {
sb.append("\nRequest Id: ").append(feedLookupResult.getRequestId());
return sb.toString();
}
+
+ public static String getString(InstanceDependencyResult dependencyResult) {
+ StringBuilder sb = new StringBuilder();
+ String results = dependencyResult.toString();
+ if (StringUtils.isEmpty(results)) {
+ sb.append("No dependencies found!");
+ } else {
+ sb.append(results);
+ }
+ sb.append("\n\nResponse: ").append(dependencyResult.getMessage());
+ sb.append("\nRequest Id: ").append(dependencyResult.getRequestId());
+ return sb.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/client/src/main/java/org/apache/falcon/cli/FalconCLI.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/falcon/cli/FalconCLI.java b/client/src/main/java/org/apache/falcon/cli/FalconCLI.java
index a5e3728..f169917 100644
--- a/client/src/main/java/org/apache/falcon/cli/FalconCLI.java
+++ b/client/src/main/java/org/apache/falcon/cli/FalconCLI.java
@@ -18,7 +18,6 @@
package org.apache.falcon.cli;
-import org.apache.falcon.ResponseHelper;
import com.sun.jersey.api.client.ClientHandlerException;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
@@ -28,12 +27,14 @@ import org.apache.commons.cli.ParseException;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.falcon.LifeCycle;
+import org.apache.falcon.ResponseHelper;
import org.apache.falcon.client.FalconCLIException;
import org.apache.falcon.client.FalconClient;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.resource.EntityList;
import org.apache.falcon.resource.FeedLookupResult;
+import org.apache.falcon.resource.InstanceDependencyResult;
import org.apache.falcon.resource.InstancesResult;
import java.io.IOException;
@@ -102,6 +103,7 @@ public class FalconCLI {
public static final String FORCE_RERUN_FLAG = "force";
public static final String INSTANCE_CMD = "instance";
+ public static final String INSTANCE_TIME_OPT = "instanceTime";
public static final String START_OPT = "start";
public static final String END_OPT = "end";
public static final String RUNNING_OPT = "running";
@@ -229,6 +231,7 @@ public class FalconCLI {
String result;
String type = commandLine.getOptionValue(ENTITY_TYPE_OPT);
String entity = commandLine.getOptionValue(ENTITY_NAME_OPT);
+ String instanceTime = commandLine.getOptionValue(INSTANCE_TIME_OPT);
String start = commandLine.getOptionValue(START_OPT);
String end = commandLine.getOptionValue(END_OPT);
String filePath = commandLine.getOptionValue(FILE_PATH_OPT);
@@ -250,7 +253,12 @@ public class FalconCLI {
validateInstanceCommands(optionsList, entity, type, colo);
- if (optionsList.contains(RUNNING_OPT)) {
+ if (optionsList.contains(DEPENDENCY_OPT)) {
+ validateNotEmpty(instanceTime, INSTANCE_TIME_OPT);
+ InstanceDependencyResult response = client.getInstanceDependencies(type, entity, instanceTime, colo);
+ result = ResponseHelper.getString(response);
+
+ } else if (optionsList.contains(RUNNING_OPT)) {
validateOrderBy(orderBy, instanceAction);
validateFilterBy(filterBy, instanceAction);
result =
@@ -785,6 +793,11 @@ public class FalconCLI {
false,
"Displays feed listing and their status between a start and end time range.");
+ Option dependency = new Option(
+ DEPENDENCY_OPT,
+ false,
+ "Displays dependent instances for a specified instance.");
+
OptionGroup group = new OptionGroup();
group.addOption(running);
group.addOption(list);
@@ -798,6 +811,7 @@ public class FalconCLI {
group.addOption(logs);
group.addOption(params);
group.addOption(listing);
+ group.addOption(dependency);
Option url = new Option(URL_OPTION, true, "Falcon URL");
Option start = new Option(START_OPT, true,
@@ -843,6 +857,8 @@ public class FalconCLI {
Option forceRerun = new Option(FORCE_RERUN_FLAG, false,
"Flag to forcefully rerun entire workflow of an instance");
+ Option instanceTime = new Option(INSTANCE_TIME_OPT, true, "Time for an instance");
+
instanceOptions.addOption(url);
instanceOptions.addOptionGroup(group);
instanceOptions.addOption(start);
@@ -861,6 +877,7 @@ public class FalconCLI {
instanceOptions.addOption(sortOrder);
instanceOptions.addOption(numResults);
instanceOptions.addOption(forceRerun);
+ instanceOptions.addOption(instanceTime);
return instanceOptions;
}
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/client/src/main/java/org/apache/falcon/client/FalconClient.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/falcon/client/FalconClient.java b/client/src/main/java/org/apache/falcon/client/FalconClient.java
index 786e0a0..20c32e4 100644
--- a/client/src/main/java/org/apache/falcon/client/FalconClient.java
+++ b/client/src/main/java/org/apache/falcon/client/FalconClient.java
@@ -37,6 +37,7 @@ import org.apache.falcon.resource.EntityList;
import org.apache.falcon.resource.EntitySummaryResult;
import org.apache.falcon.resource.FeedInstanceResult;
import org.apache.falcon.resource.FeedLookupResult;
+import org.apache.falcon.resource.InstanceDependencyResult;
import org.apache.falcon.resource.InstancesResult;
import org.apache.falcon.resource.InstancesSummaryResult;
import org.apache.falcon.resource.LineageGraphResult;
@@ -242,6 +243,7 @@ public class FalconClient {
LOG("api/instance/logs/", HttpMethod.GET, MediaType.APPLICATION_JSON),
SUMMARY("api/instance/summary/", HttpMethod.GET, MediaType.APPLICATION_JSON),
PARAMS("api/instance/params/", HttpMethod.GET, MediaType.APPLICATION_JSON),
+ DEPENDENCY("api/instance/dependencies/", HttpMethod.GET, MediaType.APPLICATION_JSON),
LISTING("api/instance/listing/", HttpMethod.GET, MediaType.APPLICATION_JSON);
private String path;
@@ -805,6 +807,22 @@ public class FalconClient {
return clientResponse;
}
+ public InstanceDependencyResult getInstanceDependencies(String entityType, String entityName, String instanceTime,
+ String colo) throws FalconCLIException {
+ checkType(entityType);
+ Instances api = Instances.DEPENDENCY;
+
+ WebResource resource = service.path(api.path).path(entityType).path(entityName);
+ resource = resource.queryParam("instanceTime", instanceTime);
+ resource = resource.queryParam("colo", colo);
+ ClientResponse clientResponse = resource
+ .header("Cookie", AUTH_COOKIE_EQ + authenticationToken)
+ .accept(api.mimeType)
+ .method(api.method, ClientResponse.class);
+ checkIfSuccessful(clientResponse);
+ return clientResponse.getEntity(InstanceDependencyResult.class);
+ }
+
//RESUME CHECKSTYLE CHECK VisibilityModifierCheck
private void checkLifeCycleOption(List<LifeCycle> lifeCycles, String type) throws FalconCLIException {
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/client/src/main/java/org/apache/falcon/resource/EntityList.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/falcon/resource/EntityList.java b/client/src/main/java/org/apache/falcon/resource/EntityList.java
index ee33234..6e132f0 100644
--- a/client/src/main/java/org/apache/falcon/resource/EntityList.java
+++ b/client/src/main/java/org/apache/falcon/resource/EntityList.java
@@ -36,6 +36,8 @@ import java.util.List;
@XmlAccessorType(XmlAccessType.FIELD)
@edu.umd.cs.findbugs.annotations.SuppressWarnings({"EI_EXPOSE_REP", "EI_EXPOSE_REP2"})
public class EntityList {
+ public static final String INPUT_TAG = "Input";
+ public static final String OUTPUT_TAG = "Output";
@XmlElement
private int totalResults;
@@ -184,14 +186,14 @@ public class EntityList {
if (process.getInputs() != null) {
for (Input i : process.getInputs().getInputs()) {
if (i.getFeed().equals(entityNameToMatch)) {
- tagList.add("Input");
+ tagList.add(INPUT_TAG);
}
}
}
if (process.getOutputs() != null) {
for (Output o : process.getOutputs().getOutputs()) {
if (o.getFeed().equals(entityNameToMatch)) {
- tagList.add("Output");
+ tagList.add(OUTPUT_TAG);
}
}
}
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/client/src/main/java/org/apache/falcon/resource/InstanceDependencyResult.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/falcon/resource/InstanceDependencyResult.java b/client/src/main/java/org/apache/falcon/resource/InstanceDependencyResult.java
new file mode 100644
index 0000000..0751f12
--- /dev/null
+++ b/client/src/main/java/org/apache/falcon/resource/InstanceDependencyResult.java
@@ -0,0 +1,86 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.falcon.resource;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlRootElement;
+
+/**
+ * Entity list used for marshalling / unmarshalling with REST calls.
+ */
+@XmlRootElement(name = "dependents")
+@XmlAccessorType(XmlAccessType.FIELD)
+@edu.umd.cs.findbugs.annotations.SuppressWarnings({"EI_EXPOSE_REP", "EI_EXPOSE_REP2"})
+public class InstanceDependencyResult extends APIResult {
+
+ @XmlElement(name = "dependencies")
+ private SchedulableEntityInstance[] dependencies;
+
+ //For JAXB
+ private InstanceDependencyResult() {
+ super();
+ }
+
+ public InstanceDependencyResult(Status status, String message) {
+ super(status, message);
+ }
+
+ public SchedulableEntityInstance[] getDependencies() {
+ return dependencies;
+ }
+
+ public void setDependencies(SchedulableEntityInstance[] dependencies) {
+ this.dependencies = dependencies;
+ }
+
+
+ @Override
+ public Object[] getCollection() {
+ return getDependencies();
+ }
+
+ @Override
+ public void setCollection(Object[] items) {
+ if (items == null) {
+ setDependencies(new SchedulableEntityInstance[0]);
+ } else {
+ SchedulableEntityInstance[] newInstances = new SchedulableEntityInstance[items.length];
+ for (int index = 0; index < items.length; index++) {
+ newInstances[index] = (SchedulableEntityInstance)items[index];
+ }
+ setDependencies(newInstances);
+ }
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder buffer = new StringBuilder();
+ if (dependencies != null) {
+ for (SchedulableEntityInstance element : dependencies) {
+ buffer.append(element.toString());
+ buffer.append("\n");
+ }
+ }
+ return buffer.toString();
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/client/src/main/java/org/apache/falcon/resource/SchedulableEntityInstance.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/falcon/resource/SchedulableEntityInstance.java b/client/src/main/java/org/apache/falcon/resource/SchedulableEntityInstance.java
new file mode 100644
index 0000000..2a7ecdb
--- /dev/null
+++ b/client/src/main/java/org/apache/falcon/resource/SchedulableEntityInstance.java
@@ -0,0 +1,155 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.falcon.resource;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.falcon.entity.v0.EntityType;
+import org.apache.falcon.entity.v0.SchemaHelper;
+
+import java.util.Date;
+
+/**
+ * Instance of a Schedulable Entity (Feed/Process).
+ */
+public class SchedulableEntityInstance {
+
+ public static final String INPUT = "Input";
+ public static final String OUTPUT = "Output";
+
+ private String entityName;
+
+ private String cluster;
+
+ private Date instanceTime;
+
+ private EntityType entityType;
+
+ private String tag;
+
+ //for JAXB
+ private SchedulableEntityInstance() {
+
+ }
+
+ public SchedulableEntityInstance(String entityName, String cluster, Date instanceTime, EntityType type) {
+ this.entityName = entityName;
+ this.cluster = cluster;
+ this.entityType = type;
+ if (instanceTime != null) {
+ this.instanceTime = new Date(instanceTime.getTime());
+ }
+ }
+
+ public String getTag() {
+ return tag;
+ }
+
+ public void setTag(String tag) {
+ this.tag = tag;
+ }
+
+ public String getEntityName() {
+ return entityName;
+ }
+
+ public void setEntityName(String entityName) {
+ this.entityName = entityName;
+ }
+
+ public String getCluster() {
+ return cluster;
+ }
+
+ public void setCluster(String cluster) {
+ this.cluster = cluster;
+ }
+
+ public EntityType getEntityType() {
+ return entityType;
+ }
+
+ public void setEntityType(EntityType entityType) {
+ this.entityType = entityType;
+ }
+
+ public Date getInstanceTime() {
+ return new Date(instanceTime.getTime());
+ }
+
+ public void setInstanceTime(Date instanceTime) {
+ this.instanceTime = new Date(instanceTime.getTime());
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("name: " + entityName
+ + ", type: " + entityType
+ + ", cluster: " + cluster
+ + ", instanceTime: " + SchemaHelper.formatDateUTC(instanceTime));
+ sb.append(", tag: " + ((tag != null) ? tag : ""));
+ return sb.toString();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ SchedulableEntityInstance that = (SchedulableEntityInstance) o;
+
+ if (instanceTime == null ? that.instanceTime != null : !instanceTime.equals(that.instanceTime)) {
+ return false;
+ }
+
+ if (!entityType.equals(that.entityType)) {
+ return false;
+ }
+
+ if (!StringUtils.equals(entityName, that.entityName)) {
+ return false;
+ }
+
+ if (!StringUtils.equals(cluster, that.cluster)) {
+ return false;
+ }
+
+ if (!StringUtils.equals(tag, that.tag)) {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = instanceTime.hashCode();
+ result = 31 * result + entityName.hashCode();
+ result = 31 * result + entityType.hashCode();
+ result = 31 * result + cluster.hashCode();
+ if (tag != null) {
+ result = 31 * result + tag.hashCode();
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/common/src/main/java/org/apache/falcon/entity/EntityUtil.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/EntityUtil.java b/common/src/main/java/org/apache/falcon/entity/EntityUtil.java
index 26d3da2..7ebf39e 100644
--- a/common/src/main/java/org/apache/falcon/entity/EntityUtil.java
+++ b/common/src/main/java/org/apache/falcon/entity/EntityUtil.java
@@ -28,6 +28,7 @@ import org.apache.falcon.Tag;
import org.apache.falcon.entity.WorkflowNameBuilder.WorkflowName;
import org.apache.falcon.entity.store.ConfigurationStore;
import org.apache.falcon.entity.v0.Entity;
+import org.apache.falcon.entity.v0.EntityGraph;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.Frequency;
import org.apache.falcon.entity.v0.SchemaHelper;
@@ -35,9 +36,13 @@ import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.cluster.ClusterLocationType;
import org.apache.falcon.entity.v0.feed.ClusterType;
import org.apache.falcon.entity.v0.feed.Feed;
-import org.apache.falcon.entity.v0.process.*;
+import org.apache.falcon.entity.v0.process.LateInput;
+import org.apache.falcon.entity.v0.process.LateProcess;
+import org.apache.falcon.entity.v0.process.PolicyType;
import org.apache.falcon.entity.v0.process.Process;
+import org.apache.falcon.entity.v0.process.Retry;
import org.apache.falcon.hadoop.HadoopClientFactory;
+import org.apache.falcon.resource.EntityList;
import org.apache.falcon.util.DeploymentUtil;
import org.apache.falcon.util.RuntimeProperties;
import org.apache.hadoop.fs.FileStatus;
@@ -53,7 +58,19 @@ import java.lang.reflect.Method;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Calendar;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TimeZone;
/**
* Helper to get entity object.
@@ -65,6 +82,7 @@ public final class EntityUtil {
private static final long HOUR_IN_MS = 60 * MINUTE_IN_MS;
private static final long DAY_IN_MS = 24 * HOUR_IN_MS;
private static final long MONTH_IN_MS = 31 * DAY_IN_MS;
+ private static final long ONE_MS = 1;
public static final String SUCCEEDED_FILE_NAME = "_SUCCESS";
@@ -244,8 +262,8 @@ public final class EntityUtil {
return feed.getTimezone();
}
- public static Date getNextStartTime(Date startTime, Frequency frequency, TimeZone timezone, Date now) {
- if (startTime.after(now)) {
+ public static Date getNextStartTime(Date startTime, Frequency frequency, TimeZone timezone, Date referenceTime) {
+ if (startTime.after(referenceTime)) {
return startTime;
}
@@ -255,16 +273,16 @@ public final class EntityUtil {
int count = 0;
switch (frequency.getTimeUnit()) {
case months:
- count = (int) ((now.getTime() - startTime.getTime()) / MONTH_IN_MS);
+ count = (int) ((referenceTime.getTime() - startTime.getTime()) / MONTH_IN_MS);
break;
case days:
- count = (int) ((now.getTime() - startTime.getTime()) / DAY_IN_MS);
+ count = (int) ((referenceTime.getTime() - startTime.getTime()) / DAY_IN_MS);
break;
case hours:
- count = (int) ((now.getTime() - startTime.getTime()) / HOUR_IN_MS);
+ count = (int) ((referenceTime.getTime() - startTime.getTime()) / HOUR_IN_MS);
break;
case minutes:
- count = (int) ((now.getTime() - startTime.getTime()) / MINUTE_IN_MS);
+ count = (int) ((referenceTime.getTime() - startTime.getTime()) / MINUTE_IN_MS);
break;
default:
}
@@ -273,7 +291,7 @@ public final class EntityUtil {
if (count > 2) {
startCal.add(frequency.getTimeUnit().getCalendarUnit(), ((count - 2) / freq) * freq);
}
- while (startCal.getTime().before(now)) {
+ while (startCal.getTime().before(referenceTime)) {
startCal.add(frequency.getTimeUnit().getCalendarUnit(), freq);
}
return startCal.getTime();
@@ -747,6 +765,12 @@ public final class EntityUtil {
return pipelines;
}
+ public static EntityList getEntityDependencies(Entity entity) throws FalconException {
+ Set<Entity> dependents = EntityGraph.get().getDependents(entity);
+ Entity[] dependentEntities = dependents.toArray(new Entity[dependents.size()]);
+ return new EntityList(dependentEntities, entity);
+ }
+
public static Pair<Date, Date> getEntityStartEndDates(Entity entityObject) {
Set<String> clusters = EntityUtil.getClustersDefined(entityObject);
Pair<Date, String> clusterMinStartDate = null;
@@ -761,4 +785,104 @@ public final class EntityUtil {
}
return new Pair<Date, Date>(clusterMinStartDate.first, clusterMaxEndDate.first);
}
+
+ /**
+ * Returns the previous instance(before or on) for a given referenceTime
+ *
+ * Example: For a feed in "UTC" with startDate "2014-01-01 00:00" and frequency of "days(1)" a referenceTime
+ * of "2015-01-01 00:00" will return "2015-01-01 00:00".
+ *
+ * Similarly for the above feed if we give a reference Time of "2015-01-01 04:00" will also result in
+ * "2015-01-01 00:00"
+ *
+ * @param startTime start time of the entity
+ * @param frequency frequency of the entity
+ * @param tz timezone of the entity
+ * @param referenceTime time before which the instanceTime is desired
+ * @return instance(before or on) the referenceTime
+ */
+ public static Date getPreviousInstanceTime(Date startTime, Frequency frequency, TimeZone tz, Date referenceTime) {
+ if (tz == null) {
+ tz = TimeZone.getTimeZone("UTC");
+ }
+ Calendar insCal = Calendar.getInstance(tz);
+ insCal.setTime(startTime);
+
+ int instanceCount = getInstanceSequence(startTime, frequency, tz, referenceTime);
+ final int freq = frequency.getFrequencyAsInt() * instanceCount;
+ insCal.add(frequency.getTimeUnit().getCalendarUnit(), freq);
+
+ while (insCal.getTime().after(referenceTime)) {
+ insCal.add(frequency.getTimeUnit().getCalendarUnit(), -1);
+ }
+ return insCal.getTime();
+ }
+
+ /**
+ * Find the times at which the given entity will run in a given time range.
+ * <p/>
+ * Both start and end Date are inclusive.
+ *
+ * @param entity feed or process entity whose instance times are to be found
+ * @param clusterName name of the cluster
+ * @param startRange start time for the input range
+ * @param endRange end time for the input range
+ * @return List of instance times at which the entity will run in the given time range
+ */
+ public static List<Date> getEntityInstanceTimes(Entity entity, String clusterName, Date startRange, Date endRange) {
+ Date start = null;
+ switch (entity.getEntityType()) {
+
+ case FEED:
+ Feed feed = (Feed) entity;
+ start = FeedHelper.getCluster(feed, clusterName).getValidity().getStart();
+ return getInstanceTimes(start, feed.getFrequency(), feed.getTimezone(),
+ startRange, endRange);
+
+ case PROCESS:
+ Process process = (Process) entity;
+ start = ProcessHelper.getCluster(process, clusterName).getValidity().getStart();
+ return getInstanceTimes(start, process.getFrequency(),
+ process.getTimezone(), startRange, endRange);
+
+ default:
+ throw new IllegalArgumentException("Unhandled type: " + entity.getEntityType());
+ }
+ }
+
+
+ /**
+ * Find instance times given first instance start time and frequency till a given end time.
+ *
+ * It finds the first valid instance time in the given time range, it then uses frequency to find next instances
+ * in the given time range.
+ *
+ * @param startTime startTime of the entity (time of first instance ever of the given entity)
+ * @param frequency frequency of the entity
+ * @param timeZone timeZone of the entity
+ * @param startRange start time for the input range of interest.
+ * @param endRange end time for the input range of interest.
+ * @return list of instance run times of the given entity in the given time range.
+ */
+ public static List<Date> getInstanceTimes(Date startTime, Frequency frequency, TimeZone timeZone,
+ Date startRange, Date endRange) {
+ List<Date> result = new LinkedList<>();
+ if (timeZone == null) {
+ timeZone = TimeZone.getTimeZone("UTC");
+ }
+
+ while(true){
+ Date nextStartTime = getNextStartTime(startTime, frequency, timeZone, startRange);
+ if (nextStartTime.before(startRange) || nextStartTime.after(endRange)){
+ break;
+ }
+
+ result.add(nextStartTime);
+ // this is required because getNextStartTime returns greater than or equal to referenceTime
+ startRange = new Date(nextStartTime.getTime() + ONE_MS); // 1 milli seconds later
+ }
+ return result;
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/FeedHelper.java b/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
index acb8598..9f4eb61 100644
--- a/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
+++ b/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
@@ -19,6 +19,7 @@
package org.apache.falcon.entity;
import org.apache.commons.lang3.StringUtils;
+
import org.apache.falcon.FalconException;
import org.apache.falcon.LifeCycle;
import org.apache.falcon.Tag;
@@ -33,16 +34,33 @@ import org.apache.falcon.entity.v0.feed.Location;
import org.apache.falcon.entity.v0.feed.LocationType;
import org.apache.falcon.entity.v0.feed.Locations;
import org.apache.falcon.entity.v0.feed.Sla;
+import org.apache.falcon.entity.v0.process.Input;
+import org.apache.falcon.entity.v0.process.Output;
+import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.expression.ExpressionHelper;
import org.apache.falcon.resource.APIResult;
+import org.apache.falcon.resource.EntityList;
import org.apache.falcon.resource.FeedInstanceResult;
+import org.apache.falcon.resource.SchedulableEntityInstance;
import org.apache.falcon.util.BuildProperties;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.URISyntaxException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TimeZone;
import java.util.regex.Matcher;
/**
@@ -50,6 +68,9 @@ import java.util.regex.Matcher;
*/
public final class FeedHelper {
+ private static final Logger LOG = LoggerFactory.getLogger(FeedHelper.class);
+ private static final int ONE_MS = 1;
+
public static final String FORMAT = "yyyyMMddHHmm";
private FeedHelper() {}
@@ -273,7 +294,7 @@ public final class FeedHelper {
public static Properties getClusterProperties(org.apache.falcon.entity.v0.cluster.Cluster cluster) {
Properties properties = new Properties();
- Map<String, String> clusterVars = new HashMap<String, String>();
+ Map<String, String> clusterVars = new HashMap<>();
clusterVars.put("colo", cluster.getColo());
clusterVars.put("name", cluster.getName());
if (cluster.getProperties() != null) {
@@ -354,7 +375,7 @@ public final class FeedHelper {
* Extracts date from the actual data path e.g., /path/2014/05/06 maps to 2014-05-06T00:00Z.
* @param instancePath - actual data path
* @param templatePath - template path from feed definition
- * @param timeZone
+ * @param timeZone timeZone
* @return date corresponding to the path
*/
//consider just the first occurrence of the pattern
@@ -364,7 +385,7 @@ public final class FeedHelper {
Calendar cal = Calendar.getInstance(timeZone);
int lastEnd = 0;
- Set<FeedDataPath.VARS> matchedVars = new HashSet<FeedDataPath.VARS>();
+ Set<FeedDataPath.VARS> matchedVars = new HashSet<>();
while (matcher.find()) {
FeedDataPath.VARS pathVar = FeedDataPath.VARS.from(matcher.group());
String pad = templatePath.substring(lastEnd, matcher.start());
@@ -415,6 +436,264 @@ public final class FeedHelper {
}
+ private static void validateFeedInstance(Feed feed, Date instanceTime,
+ org.apache.falcon.entity.v0.cluster.Cluster cluster) {
+
+ // validate the cluster
+ Cluster feedCluster = getCluster(feed, cluster.getName());
+ if (feedCluster == null) {
+ throw new IllegalArgumentException("Cluster :" + cluster.getName() + " is not a valid cluster for feed:"
+ + feed.getName());
+ }
+
+ // validate that instanceTime is in validity range
+ if (feedCluster.getValidity().getStart().after(instanceTime)
+ || feedCluster.getValidity().getEnd().before(instanceTime)) {
+ throw new IllegalArgumentException("instanceTime: " + instanceTime + " is not in validity range for"
+ + " Feed: " + feed.getName() + " on cluster:" + cluster.getName());
+ }
+
+ // validate instanceTime on basis of startTime and frequency
+ Date nextInstance = EntityUtil.getNextStartTime(feedCluster.getValidity().getStart(), feed.getFrequency(),
+ feed.getTimezone(), instanceTime);
+ if (!nextInstance.equals(instanceTime)) {
+ throw new IllegalArgumentException("instanceTime: " + instanceTime + " is not a valid instance for the "
+ + " feed: " + feed.getName() + " on cluster: " + cluster.getName()
+ + " on the basis of startDate and frequency");
+ }
+ }
+
+ /**
+ * Given a feed Instance finds the generating process instance.
+ *
+ * [process, cluster, instanceTime]
+ *
+ * If the feed is replicated, then it returns null.
+ *
+ * @param feed output feed
+ * @param feedInstanceTime instance time of the feed
+ * @return returns the instance of the process which produces the given feed
+ */
+ public static SchedulableEntityInstance getProducerInstance(Feed feed, Date feedInstanceTime,
+ org.apache.falcon.entity.v0.cluster.Cluster cluster) throws FalconException {
+
+ //validate the inputs
+ validateFeedInstance(feed, feedInstanceTime, cluster);
+
+ Process process = getProducerProcess(feed);
+ if (process != null) {
+ try {
+ Date processInstanceTime = getProducerInstanceTime(feed, feedInstanceTime, process, cluster);
+ SchedulableEntityInstance producer = new SchedulableEntityInstance(process.getName(), cluster.getName(),
+ processInstanceTime, EntityType.PROCESS);
+ producer.setTag(SchedulableEntityInstance.OUTPUT);
+ return producer;
+ } catch (FalconException e) {
+ LOG.error("Error in trying to get producer process: {}'s instance time for feed: {}'s instance: } "
+ + " on cluster:{}", process.getName(), feed.getName(), feedInstanceTime, cluster.getName());
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Given a feed find it's generating process.
+ *
+ * If no generating process is found it returns null.
+ * @param feed output feed
+ * @return Process which produces the given feed.
+ */
+ public static Process getProducerProcess(Feed feed) throws FalconException {
+
+ EntityList dependencies = EntityUtil.getEntityDependencies(feed);
+
+ for (EntityList.EntityElement e : dependencies.getElements()) {
+ if (e.tag.contains(EntityList.OUTPUT_TAG)) {
+ return EntityUtil.getEntity(EntityType.PROCESS, e.name);
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Find the producerInstance which will generate the given feedInstance.
+ *
+ * @param feed output feed
+ * @param feedInstanceTime instance time of the output feed
+ * @param producer producer process
+ * @return time of the producer instance which will produce the given feed instance.
+ */
+ private static Date getProducerInstanceTime(Feed feed, Date feedInstanceTime, Process producer,
+ org.apache.falcon.entity.v0.cluster.Cluster cluster) throws FalconException {
+
+ String clusterName = cluster.getName();
+ Cluster feedCluster = getCluster(feed, clusterName);
+ org.apache.falcon.entity.v0.process.Cluster processCluster = ProcessHelper.getCluster(producer, clusterName);
+ Date producerStartDate = processCluster.getValidity().getStart();
+
+ // read the process definition and find the relative time difference between process and output feed
+ // if output process instance time is now then output FeedInstance time is x
+ String outputInstance = null;
+ for (Output op : producer.getOutputs().getOutputs()) {
+ if (StringUtils.equals(feed.getName(), op.getFeed())) {
+ outputInstance = op.getInstance();
+ }
+ }
+
+ ExpressionHelper.setReferenceDate(producerStartDate);
+ ExpressionHelper evaluator = ExpressionHelper.get();
+ // producerInstance = feedInstanceTime + (difference between producer process and feed)
+ // the feedInstance before or equal to this time is the required one
+ Date relativeFeedInstance = evaluator.evaluate(outputInstance, Date.class);
+ Date feedInstanceActual = EntityUtil.getPreviousInstanceTime(feedCluster.getValidity().getStart(),
+ feed.getFrequency(), feed.getTimezone(), relativeFeedInstance);
+ Long producerInstanceTime = feedInstanceTime.getTime() + (producerStartDate.getTime()
+ - feedInstanceActual.getTime());
+ Date producerInstance = new Date(producerInstanceTime);
+
+ //validate that the producerInstance is in the validity range on the provided cluster
+ if (producerInstance.before(processCluster.getValidity().getStart())
+ || producerInstance.after(processCluster.getValidity().getEnd())) {
+ throw new IllegalArgumentException("Instance time provided: " + feedInstanceTime
+ + " for feed " + feed.getName()
+ + " is outside the range of instances produced by the producer process: " + producer.getName()
+ + " in it's validity range on provided cluster: " + cluster.getName());
+ }
+ return producerInstance;
+ }
+
+
+ public static Set<SchedulableEntityInstance> getConsumerInstances(Feed feed, Date feedInstanceTime,
+ org.apache.falcon.entity.v0.cluster.Cluster cluster) throws FalconException {
+
+ Set<SchedulableEntityInstance> result = new HashSet<>();
+ // validate that the feed has this cluster & validate that the instanceTime is a valid instanceTime
+ validateFeedInstance(feed, feedInstanceTime, cluster);
+
+ Set<Process> consumers = getConsumerProcesses(feed);
+ for (Process p : consumers) {
+ Set<Date> consumerInstanceTimes = getConsumerProcessInstanceTimes(feed, feedInstanceTime, p, cluster);
+ for (Date date : consumerInstanceTimes) {
+ SchedulableEntityInstance in = new SchedulableEntityInstance(p.getName(), cluster.getName(), date,
+ EntityType.PROCESS);
+ in.setTag(SchedulableEntityInstance.INPUT);
+ result.add(in);
+ }
+ }
+ return result;
+ }
+
+
+ /**
+ * Returns the consumer processes for a given feed if any, null otherwise.
+ *
+ * @param feed input feed
+ * @return the set of processes which use the given feed as input, empty set if no consumers.
+ */
+ public static Set<Process> getConsumerProcesses(Feed feed) throws FalconException {
+ Set<Process> result = new HashSet<>();
+ EntityList dependencies = EntityUtil.getEntityDependencies(feed);
+
+ for (EntityList.EntityElement e : dependencies.getElements()) {
+ if (e.tag.contains(EntityList.INPUT_TAG)) {
+ Process consumer = EntityUtil.getEntity(EntityType.PROCESS, e.name);
+ result.add(consumer);
+ }
+ }
+ return result;
+ }
+
+ // return all instances of a process which will consume the given feed instance
+ private static Set<Date> getConsumerProcessInstanceTimes(Feed feed, Date feedInstancetime, Process consumer,
+ org.apache.falcon.entity.v0.cluster.Cluster cluster) throws FalconException {
+
+ Set<Date> result = new HashSet<>();
+ // find relevant cluster for the process
+ org.apache.falcon.entity.v0.process.Cluster processCluster =
+ ProcessHelper.getCluster(consumer, cluster.getName());
+ if (processCluster == null) {
+ throw new IllegalArgumentException("Cluster is not valid for process");
+ }
+ Date processStartDate = processCluster.getValidity().getStart();
+ Cluster feedCluster = getCluster(feed, cluster.getName());
+ Date feedStartDate = feedCluster.getValidity().getStart();
+
+ // find all corresponding Inputs as a process may refer same feed multiple times
+ List<Input> inputFeeds = new ArrayList<>();
+ if (consumer.getInputs() != null && consumer.getInputs().getInputs() != null) {
+ for (Input input : consumer.getInputs().getInputs()) {
+ if (StringUtils.equals(input.getFeed(), feed.getName())) {
+ inputFeeds.add(input);
+ }
+ }
+ }
+
+ // for each input corresponding to given feed, find corresponding consumer instances
+ for (Input in : inputFeeds) {
+ /* Algorithm for finding a consumer instance for an input feed instance
+ Step 1. Find one instance which will consume the given feed instance.
+ a. take process start date and find last input feed instance time. In this step take care of
+ frequencies being out of sync.
+ b. using the above find the time difference between the process instance and feed instance.
+ c. Adding the above time difference to given feed instance for which we want to find the consumer
+ instances we will get one consumer process instance.
+ Step 2. Keep checking for next instances of process till they consume the given feed Instance.
+ Step 3. Similarly check for all previous instances of process till they consume the given feed instance.
+ */
+
+ // Step 1.a & 1.b
+ ExpressionHelper.setReferenceDate(processStartDate);
+ ExpressionHelper evaluator = ExpressionHelper.get();
+ Date startRelative = evaluator.evaluate(in.getStart(), Date.class);
+ Date startTimeActual = EntityUtil.getNextStartTime(feedStartDate,
+ feed.getFrequency(), feed.getTimezone(), startRelative);
+ Long offset = processStartDate.getTime() - startTimeActual.getTime();
+
+ // Step 1.c
+ Date processInstanceStartRelative = new Date(feedInstancetime.getTime() + offset);
+ Date processInstanceStartActual = EntityUtil.getPreviousInstanceTime(processStartDate,
+ consumer.getFrequency(), consumer.getTimezone(), processInstanceStartRelative);
+
+
+ // Step 2.
+ Date currentInstance = processInstanceStartActual;
+ while (true) {
+ Date nextConsumerInstance = EntityUtil.getNextStartTime(processStartDate,
+ consumer.getFrequency(), consumer.getTimezone(), currentInstance);
+
+ ExpressionHelper.setReferenceDate(nextConsumerInstance);
+ evaluator = ExpressionHelper.get();
+ Long rangeStart = evaluator.evaluate(in.getStart(), Date.class).getTime();
+ Long rangeEnd = evaluator.evaluate(in.getEnd(), Date.class).getTime();
+ if (rangeStart <= feedInstancetime.getTime() && feedInstancetime.getTime() < rangeEnd) {
+ result.add(nextConsumerInstance);
+ } else {
+ break;
+ }
+ currentInstance = new Date(nextConsumerInstance.getTime() + ONE_MS);
+ }
+
+ // Step 3.
+ currentInstance = processInstanceStartActual;
+ while (true) {
+ Date nextConsumerInstance = EntityUtil.getPreviousInstanceTime(processStartDate,
+ consumer.getFrequency(), consumer.getTimezone(), currentInstance);
+
+ ExpressionHelper.setReferenceDate(nextConsumerInstance);
+ evaluator = ExpressionHelper.get();
+ Long rangeStart = evaluator.evaluate(in.getStart(), Date.class).getTime();
+ Long rangeEnd = evaluator.evaluate(in.getEnd(), Date.class).getTime();
+ if (rangeStart <= feedInstancetime.getTime() && feedInstancetime.getTime() < rangeEnd) {
+ result.add(nextConsumerInstance);
+ } else {
+ break;
+ }
+ currentInstance = new Date(nextConsumerInstance.getTime() - ONE_MS);
+ }
+ }
+ return result;
+ }
+
public static FeedInstanceResult getFeedInstanceListing(Entity entityObject,
Date start, Date end) throws FalconException {
Set<String> clusters = EntityUtil.getClustersDefinedInColos(entityObject);
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java b/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java
index 29aefa0..fe78bc8 100644
--- a/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java
+++ b/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java
@@ -20,12 +20,20 @@ package org.apache.falcon.entity;
import org.apache.commons.lang3.StringUtils;
import org.apache.falcon.FalconException;
+import org.apache.falcon.entity.store.ConfigurationStore;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.feed.Feed;
import org.apache.falcon.entity.v0.process.Cluster;
import org.apache.falcon.entity.v0.process.Input;
import org.apache.falcon.entity.v0.process.Output;
import org.apache.falcon.entity.v0.process.Process;
+import org.apache.falcon.expression.ExpressionHelper;
+import org.apache.falcon.resource.SchedulableEntityInstance;
+
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
/**
* Helper methods for accessing process members.
@@ -34,6 +42,7 @@ public final class ProcessHelper {
private ProcessHelper() {}
+
public static Cluster getCluster(Process process, String clusterName) {
for (Cluster cluster : process.getClusters().getClusters()) {
if (cluster.getName().equals(clusterName)) {
@@ -77,4 +86,103 @@ public final class ProcessHelper {
return storageType;
}
+
+ private static void validateProcessInstance(Process process, Date instanceTime,
+ org.apache.falcon.entity.v0.cluster.Cluster cluster) {
+ //validate the cluster
+ Cluster processCluster = getCluster(process, cluster.getName());
+ if (processCluster == null) {
+ throw new IllegalArgumentException("Cluster provided: " + cluster.getName()
+ + " is not a valid cluster for the process: " + process.getName());
+ }
+
+ // check if instanceTime is in validity range
+ if (instanceTime.before(processCluster.getValidity().getStart())
+ || instanceTime.after(processCluster.getValidity().getEnd())) {
+ throw new IllegalArgumentException("Instance time provided: " + instanceTime
+ + " is not in validity range of process: " + process.getName()
+ + "on cluster: " + cluster.getName());
+ }
+
+ // check instanceTime is valid on the basis of startTime and frequency
+ Date nextInstance = EntityUtil.getNextStartTime(processCluster.getValidity().getStart(),
+ process.getFrequency(), process.getTimezone(), instanceTime);
+ if (!nextInstance.equals(instanceTime)) {
+ throw new IllegalArgumentException("Instance time provided: " + instanceTime
+ + " for process: " + process.getName() + " is not a valid instance time on cluster: "
+ + cluster.getName() + " on the basis of startDate and frequency");
+ }
+ }
+
+ /**
+ * Given a process instance, returns the feed instances which are used as input for this process instance.
+ *
+ * @param process given process
+ * @param instanceTime nominal time of the process instance
+ * @param cluster - cluster for the process instance
+ * @param allowOptionalFeeds switch to indicate whether optional feeds should be considered in input feeds.
+ * @return Set of input feed instances which are consumed by the given process instance.
+ * @throws org.apache.falcon.FalconException
+ */
+ public static Set<SchedulableEntityInstance> getInputFeedInstances(Process process, Date instanceTime,
+ org.apache.falcon.entity.v0.cluster.Cluster cluster, boolean allowOptionalFeeds) throws FalconException {
+
+ // validate the inputs
+ validateProcessInstance(process, instanceTime, cluster);
+
+ Set<SchedulableEntityInstance> result = new HashSet<>();
+ if (process.getInputs() != null) {
+ ConfigurationStore store = ConfigurationStore.get();
+ for (Input i : process.getInputs().getInputs()) {
+ if (i.isOptional() && !allowOptionalFeeds) {
+ continue;
+ }
+ Feed feed = store.get(EntityType.FEED, i.getFeed());
+ // inputStart is process instance time + (now - startTime)
+ ExpressionHelper evaluator = ExpressionHelper.get();
+ ExpressionHelper.setReferenceDate(instanceTime);
+ Date inputInstanceStartDate = evaluator.evaluate(i.getStart(), Date.class);
+ Date inputInstanceEndDate = evaluator.evaluate(i.getEnd(), Date.class);
+ List<Date> instanceTimes = EntityUtil.getEntityInstanceTimes(feed, cluster.getName(),
+ inputInstanceStartDate, inputInstanceEndDate);
+ SchedulableEntityInstance instance;
+ for (Date time : instanceTimes) {
+ instance = new SchedulableEntityInstance(feed.getName(), cluster.getName(), time, EntityType.FEED);
+ instance.setTag(SchedulableEntityInstance.INPUT);
+ result.add(instance);
+ }
+ }
+ }
+ return result;
+ }
+
+ public static Set<SchedulableEntityInstance> getOutputFeedInstances(Process process, Date instanceTime,
+ org.apache.falcon.entity.v0.cluster.Cluster cluster) throws FalconException {
+ Set<SchedulableEntityInstance> result = new HashSet<>();
+
+ // validate the inputs
+ validateProcessInstance(process, instanceTime, cluster);
+
+ if (process.getOutputs() != null && process.getOutputs().getOutputs() != null) {
+
+ ExpressionHelper.setReferenceDate(instanceTime);
+ ExpressionHelper evaluator = ExpressionHelper.get();
+ SchedulableEntityInstance candidate;
+ ConfigurationStore store = ConfigurationStore.get();
+ for (Output output : process.getOutputs().getOutputs()) {
+
+ Date outputInstance = evaluator.evaluate(output.getInstance(), Date.class);
+ // find the feed
+ Feed feed = store.get(EntityType.FEED, output.getFeed());
+ org.apache.falcon.entity.v0.feed.Cluster fCluster = FeedHelper.getCluster(feed, cluster.getName());
+ outputInstance = EntityUtil.getNextStartTime(fCluster.getValidity().getStart(), feed.getFrequency(),
+ feed.getTimezone(), outputInstance);
+ candidate = new SchedulableEntityInstance(output.getFeed(), cluster.getName(), outputInstance,
+ EntityType.FEED);
+ candidate.setTag(SchedulableEntityInstance.OUTPUT);
+ result.add(candidate);
+ }
+ }
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/common/src/test/java/org/apache/falcon/entity/FeedHelperTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/entity/FeedHelperTest.java b/common/src/test/java/org/apache/falcon/entity/FeedHelperTest.java
index 266d029..f70edfb 100644
--- a/common/src/test/java/org/apache/falcon/entity/FeedHelperTest.java
+++ b/common/src/test/java/org/apache/falcon/entity/FeedHelperTest.java
@@ -18,24 +18,58 @@
package org.apache.falcon.entity;
+import org.apache.falcon.FalconException;
+import org.apache.falcon.entity.store.ConfigurationStore;
+import org.apache.falcon.entity.v0.EntityType;
+import org.apache.falcon.entity.v0.Frequency;
import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.cluster.Properties;
import org.apache.falcon.entity.v0.cluster.Property;
-import org.apache.falcon.entity.v0.feed.*;
+import org.apache.falcon.entity.v0.feed.Clusters;
+import org.apache.falcon.entity.v0.feed.Feed;
+import org.apache.falcon.entity.v0.feed.Location;
+import org.apache.falcon.entity.v0.feed.LocationType;
+import org.apache.falcon.entity.v0.feed.Locations;
+import org.apache.falcon.entity.v0.feed.Validity;
+import org.apache.falcon.entity.v0.process.Input;
+import org.apache.falcon.entity.v0.process.Inputs;
+import org.apache.falcon.entity.v0.process.Output;
+import org.apache.falcon.entity.v0.process.Outputs;
+import org.apache.falcon.entity.v0.process.Process;
+import org.apache.falcon.resource.SchedulableEntityInstance;
import org.apache.hadoop.fs.Path;
import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import java.text.DateFormat;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
import java.util.Date;
+import java.util.HashSet;
+import java.util.Set;
import java.util.TimeZone;
/**
* Test for feed helper methods.
*/
-public class FeedHelperTest {
- public static final TimeZone UTC = TimeZone.getTimeZone("UTC");
+public class FeedHelperTest extends AbstractTestBase {
+ private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
+ private ConfigurationStore store;
+
+ @BeforeClass
+ public void init() throws Exception {
+ initConfigStore();
+ }
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ cleanupStore();
+ store = getStore();
+ }
@Test
public void testPartitionExpression() {
@@ -107,4 +141,334 @@ public class FeedHelperTest {
locations.getLocations());
Assert.assertEquals(FeedHelper.getLocation(feed, cluster, LocationType.DATA), location2);
}
+
+ @Test
+ public void testGetProducerProcessWithOffset() throws FalconException, ParseException {
+ //create a feed, submit it, test that ProducerProcess is null
+
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "days(1)", "2011-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Assert.assertNull(FeedHelper.getProducerProcess(feed));
+
+ // create it's producer process submit it, test it's ProducerProcess
+ Process process = prepareProcess(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Outputs outputs = new Outputs();
+ Output outFeed = new Output();
+ outFeed.setName("outputFeed");
+ outFeed.setFeed(feed.getName());
+ outFeed.setInstance("today(0,0)");
+ outputs.getOutputs().add(outFeed);
+ process.setOutputs(outputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Assert.assertEquals(FeedHelper.getProducerProcess(feed).getName(), process.getName());
+ SchedulableEntityInstance result = FeedHelper.getProducerInstance(feed, getDate("2013-02-27 10:00 UTC"),
+ cluster);
+ SchedulableEntityInstance expected = new SchedulableEntityInstance(process.getName(), cluster.getName(),
+ getDate("2013-02-28 10:00 UTC"), EntityType.PROCESS);
+ expected.setTag(SchedulableEntityInstance.OUTPUT);
+ Assert.assertEquals(result, expected);
+ }
+
+ @Test
+ public void testGetProducerProcessForNow() throws FalconException, ParseException {
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "days(1)", "2011-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Assert.assertNull(FeedHelper.getProducerProcess(feed));
+
+ // create it's producer process submit it, test it's ProducerProcess
+ Process process = prepareProcess(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Outputs outputs = new Outputs();
+ Output outFeed = new Output();
+ outFeed.setName("outputFeed");
+ outFeed.setFeed(feed.getName());
+ outFeed.setInstance("now(0,0)");
+ outputs.getOutputs().add(outFeed);
+ process.setOutputs(outputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Assert.assertEquals(FeedHelper.getProducerProcess(feed).getName(), process.getName());
+ SchedulableEntityInstance result = FeedHelper.getProducerInstance(feed, getDate("2013-02-28 10:00 UTC"),
+ cluster);
+ SchedulableEntityInstance expected = new SchedulableEntityInstance(process.getName(), cluster.getName(),
+ getDate("2013-02-28 10:00 UTC"), EntityType.PROCESS);
+ expected.setTag(SchedulableEntityInstance.OUTPUT);
+ Assert.assertEquals(result, expected);
+ }
+
+ @Test
+ public void testGetProducerWithNowNegativeOffset() throws FalconException, ParseException {
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "days(1)", "2011-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Assert.assertNull(FeedHelper.getProducerProcess(feed));
+
+ // create it's producer process submit it, test it's ProducerProcess
+ Process process = prepareProcess(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Outputs outputs = new Outputs();
+ Output outFeed = new Output();
+ outFeed.setName("outputFeed");
+ outFeed.setFeed(feed.getName());
+ outFeed.setInstance("now(-4,0)");
+ outputs.getOutputs().add(outFeed);
+ process.setOutputs(outputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Assert.assertEquals(FeedHelper.getProducerProcess(feed).getName(), process.getName());
+ SchedulableEntityInstance result = FeedHelper.getProducerInstance(feed, getDate("2013-02-27 10:00 UTC"),
+ cluster);
+ SchedulableEntityInstance expected = new SchedulableEntityInstance(process.getName(), cluster.getName(),
+ getDate("2013-02-28 10:00 UTC"), EntityType.PROCESS);
+ expected.setTag(SchedulableEntityInstance.OUTPUT);
+ Assert.assertEquals(result, expected);
+ }
+
+
+ @Test
+ public void testGetProducerWithNowPositiveOffset() throws FalconException, ParseException {
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "days(1)", "2011-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Assert.assertNull(FeedHelper.getProducerProcess(feed));
+
+ // create it's producer process submit it, test it's ProducerProcess
+ Process process = prepareProcess(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Outputs outputs = new Outputs();
+ Output outFeed = new Output();
+ outFeed.setName("outputFeed");
+ outFeed.setFeed(feed.getName());
+ outFeed.setInstance("now(4,0)");
+ outputs.getOutputs().add(outFeed);
+ process.setOutputs(outputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Assert.assertEquals(FeedHelper.getProducerProcess(feed).getName(), process.getName());
+ SchedulableEntityInstance result = FeedHelper.getProducerInstance(feed, getDate("2013-02-28 10:00 UTC"),
+ cluster);
+ SchedulableEntityInstance expected = new SchedulableEntityInstance(process.getName(), cluster.getName(),
+ getDate("2013-02-28 10:00 UTC"), EntityType.PROCESS);
+ expected.setTag(SchedulableEntityInstance.OUTPUT);
+ Assert.assertEquals(result, expected);
+ }
+
+ @Test
+ public void testGetProducerProcessInstance() throws FalconException, ParseException {
+ //create a feed, submit it, test that ProducerProcess is null
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "days(1)", "2011-02-28 00:00 UTC", "2016-02-28 10:00 UTC");
+
+ // create it's producer process submit it, test it's ProducerProcess
+ Process process = prepareProcess(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Outputs outputs = new Outputs();
+ Output outFeed = new Output();
+ outFeed.setName("outputFeed");
+ outFeed.setFeed(feed.getName());
+ outFeed.setInstance("today(0,0)");
+ outputs.getOutputs().add(outFeed);
+ process.setOutputs(outputs);
+ store.publish(EntityType.PROCESS, process);
+ Assert.assertEquals(FeedHelper.getProducerProcess(feed).getName(), process.getName());
+ SchedulableEntityInstance result = FeedHelper.getProducerInstance(feed, getDate("2013-02-28 00:00 UTC"),
+ cluster);
+ SchedulableEntityInstance expected = new SchedulableEntityInstance(process.getName(), cluster.getName(),
+ getDate("2013-02-28 10:00 UTC"), EntityType.PROCESS);
+ expected.setTag(SchedulableEntityInstance.OUTPUT);
+ Assert.assertEquals(result, expected);
+ }
+
+ @Test
+ public void testGetConsumerProcesses() throws FalconException, ParseException {
+ //create a feed, submit it, test that ConsumerProcesses is blank list
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+
+ //create a consumer Process and submit it, assert that this is returned in ConsumerProcesses
+ Process process = prepareProcess(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Inputs inputs = new Inputs();
+ Input inFeed = new Input();
+ inFeed.setName("outputFeed");
+ inFeed.setFeed(feed.getName());
+ inFeed.setStart("today(0,0)");
+ inFeed.setEnd("today(0,0)");
+ inputs.getInputs().add(inFeed);
+ process.setInputs(inputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Set<Process> result = FeedHelper.getConsumerProcesses(feed);
+ Assert.assertEquals(result.size(), 1);
+ Assert.assertTrue(result.contains(process));
+ }
+
+ @Test
+ public void testGetConsumerProcessInstances() throws Exception {
+ //create a feed, submit it, test that ConsumerProcesses is blank list
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "hours(1)", "2012-02-28 00:00 UTC", "2016-02-28 00:00 UTC");
+
+ //create a consumer Process and submit it, assert that this is returned in ConsumerProcesses
+ Process process = prepareProcess(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Inputs inputs = new Inputs();
+ Input inFeed = new Input();
+ inFeed.setName("inputFeed");
+ inFeed.setFeed(feed.getName());
+ inFeed.setStart("now(-4, 30)");
+ inFeed.setEnd("now(4, 30)");
+ inputs.getInputs().add(inFeed);
+ process.setInputs(inputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Set<SchedulableEntityInstance> result = FeedHelper.getConsumerInstances(feed,
+ getDate("2012-02-28 09:00 UTC"), cluster);
+ Assert.assertEquals(result.size(), 1);
+
+ Set<SchedulableEntityInstance> expected = new HashSet<>();
+ SchedulableEntityInstance ins = new SchedulableEntityInstance(process.getName(), cluster.getName(),
+ getDate("2012-02-28 10:00 UTC"), EntityType.PROCESS);
+ ins.setTag(SchedulableEntityInstance.INPUT);
+ expected.add(ins);
+ Assert.assertEquals(result, expected);
+
+ }
+
+ @Test
+ public void testGetMultipleConsumerInstances() throws Exception {
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "hours(1)", "2012-02-28 00:00 UTC", "2016-02-28 00:00 UTC");
+
+ //create a consumer Process and submit it, assert that this is returned in ConsumerProcesses
+ Process process = prepareProcess(cluster, "hours(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Inputs inputs = new Inputs();
+ Input inFeed = new Input();
+ inFeed.setName("inputFeed");
+ inFeed.setFeed(feed.getName());
+ inFeed.setStart("now(-4, 30)");
+ inFeed.setEnd("now(4, 30)");
+ inputs.getInputs().add(inFeed);
+ process.setInputs(inputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Set<SchedulableEntityInstance> result = FeedHelper.getConsumerInstances(feed,
+ getDate("2012-02-28 09:00 UTC"), cluster);
+ Assert.assertEquals(result.size(), 8);
+
+ Set<SchedulableEntityInstance> expected = new HashSet<>();
+ String[] consumers = { "2012-02-28 05:00 UTC", "2012-02-28 06:00 UTC", "2012-02-28 07:00 UTC",
+ "2012-02-28 08:00 UTC", "2012-02-28 09:00 UTC", "2012-02-28 10:00 UTC", "2012-02-28 11:00 UTC",
+ "2012-02-28 12:00 UTC", };
+ for (String d : consumers) {
+ SchedulableEntityInstance i = new SchedulableEntityInstance(process.getName(), cluster.getName(),
+ getDate(d), EntityType.PROCESS);
+ i.setTag(SchedulableEntityInstance.INPUT);
+ expected.add(i);
+ }
+ Assert.assertEquals(result, expected);
+ }
+
+ @Test
+ public void testGetConsumerWithNow() throws Exception {
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "hours(1)", "2012-02-28 00:00 UTC", "2016-02-28 00:00 UTC");
+
+ //create a consumer Process and submit it, assert that this is returned in ConsumerProcesses
+ Process process = prepareProcess(cluster, "hours(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Inputs inputs = new Inputs();
+ Input inFeed = new Input();
+ inFeed.setName("inputFeed");
+ inFeed.setFeed(feed.getName());
+ inFeed.setStart("today(0, 0)");
+ inFeed.setEnd("now(0, 0)");
+ inputs.getInputs().add(inFeed);
+ process.setInputs(inputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Set<SchedulableEntityInstance> result = FeedHelper.getConsumerInstances(feed,
+ getDate("2012-02-28 00:00 UTC"), cluster);
+ Assert.assertEquals(result.size(), 23);
+ }
+
+ @Test
+ public void testGetConsumerWithLatest() throws Exception {
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "hours(1)", "2012-02-28 00:00 UTC", "2016-02-28 00:00 UTC");
+
+ //create a consumer Process and submit it, assert that this is returned in ConsumerProcesses
+ Process process = prepareProcess(cluster, "hours(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Inputs inputs = new Inputs();
+ Input inFeed = new Input();
+ inFeed.setName("inputFeed");
+ inFeed.setFeed(feed.getName());
+ inFeed.setStart("today(0, 0)");
+ inFeed.setEnd("latest(0)");
+ inputs.getInputs().add(inFeed);
+ process.setInputs(inputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Set<SchedulableEntityInstance> result = FeedHelper.getConsumerInstances(feed,
+ getDate("2012-02-28 00:00 UTC"), cluster);
+ System.out.println("result.size() = " + result.size());
+ Assert.assertEquals(result.size(), 23);
+ }
+
+ private Validity getFeedValidity(String start, String end) throws ParseException {
+ Validity validity = new Validity();
+ validity.setStart(getDate(start));
+ validity.setEnd(getDate(end));
+ return validity;
+ }
+
+ private org.apache.falcon.entity.v0.process.Validity getProcessValidity(String start, String end) throws
+ ParseException {
+
+ org.apache.falcon.entity.v0.process.Validity validity = new org.apache.falcon.entity.v0.process.Validity();
+ validity.setStart(getDate(start));
+ validity.setEnd(getDate(end));
+ return validity;
+ }
+
+ private Date getDate(String dateString) throws ParseException {
+ DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm Z");
+ return format.parse(dateString);
+ }
+
+ private Cluster publishCluster() throws FalconException {
+ Cluster cluster = new Cluster();
+ cluster.setName("feedCluster");
+ cluster.setColo("colo");
+ store.publish(EntityType.CLUSTER, cluster);
+ return cluster;
+
+ }
+
+ private Feed publishFeed(Cluster cluster, String frequency, String start, String end)
+ throws FalconException, ParseException {
+
+ Feed feed = new Feed();
+ feed.setName("feed");
+ Frequency f = new Frequency(frequency);
+ feed.setFrequency(f);
+ feed.setTimezone(UTC);
+ Clusters fClusters = new Clusters();
+ org.apache.falcon.entity.v0.feed.Cluster fCluster = new org.apache.falcon.entity.v0.feed.Cluster();
+ fCluster.setName(cluster.getName());
+ fCluster.setValidity(getFeedValidity(start, end));
+ fClusters.getClusters().add(fCluster);
+ feed.setClusters(fClusters);
+ store.publish(EntityType.FEED, feed);
+
+ return feed;
+ }
+
+ private Process prepareProcess(Cluster cluster, String frequency, String start, String end) throws ParseException {
+ Process process = new Process();
+ process.setName("process");
+ process.setTimezone(UTC);
+ org.apache.falcon.entity.v0.process.Clusters pClusters = new org.apache.falcon.entity.v0.process.Clusters();
+ org.apache.falcon.entity.v0.process.Cluster pCluster = new org.apache.falcon.entity.v0.process.Cluster();
+ pCluster.setName(cluster.getName());
+ org.apache.falcon.entity.v0.process.Validity validity = getProcessValidity(start, end);
+ pCluster.setValidity(validity);
+ pClusters.getClusters().add(pCluster);
+ process.setClusters(pClusters);
+ Frequency f = new Frequency(frequency);
+ process.setFrequency(f);
+ return process;
+ }
}
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/common/src/test/java/org/apache/falcon/entity/ProcessHelperTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/entity/ProcessHelperTest.java b/common/src/test/java/org/apache/falcon/entity/ProcessHelperTest.java
new file mode 100644
index 0000000..0d396ae
--- /dev/null
+++ b/common/src/test/java/org/apache/falcon/entity/ProcessHelperTest.java
@@ -0,0 +1,207 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.falcon.entity;
+
+import org.apache.falcon.FalconException;
+import org.apache.falcon.entity.store.ConfigurationStore;
+import org.apache.falcon.entity.v0.EntityType;
+import org.apache.falcon.entity.v0.Frequency;
+import org.apache.falcon.entity.v0.cluster.Cluster;
+import org.apache.falcon.entity.v0.feed.Clusters;
+import org.apache.falcon.entity.v0.feed.Feed;
+import org.apache.falcon.entity.v0.process.Input;
+import org.apache.falcon.entity.v0.process.Inputs;
+import org.apache.falcon.entity.v0.process.Output;
+import org.apache.falcon.entity.v0.process.Outputs;
+import org.apache.falcon.entity.v0.process.Process;
+import org.apache.falcon.resource.SchedulableEntityInstance;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TimeZone;
+
+
+/**
+ * Tests for ProcessHelper methods.
+ */
+public class ProcessHelperTest extends AbstractTestBase {
+ private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
+ private ConfigurationStore store;
+
+ @BeforeClass
+ public void init() throws Exception {
+ initConfigStore();
+ }
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ cleanupStore();
+ store = ConfigurationStore.get();
+ }
+
+ @Test
+ public void testGetInputFeedInstances() throws FalconException, ParseException {
+ // create a process with input feeds
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "hours(1)", "2012-02-27 10:00 UTC", "2016-02-28 10:00 UTC");
+
+ // find the input Feed instances time
+ Process process = prepareProcess(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Inputs inputs = new Inputs();
+ Input input = getInput("inputFeed", feed.getName(), "today(0,-30)", "today(2,30)", false);
+ inputs.getInputs().add(input);
+ process.setInputs(inputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Date processInstanceDate = getDate("2012-02-28 10:00 UTC");
+ Set<SchedulableEntityInstance> inputFeedInstances = ProcessHelper.getInputFeedInstances(process,
+ processInstanceDate, cluster, false);
+ Assert.assertEquals(inputFeedInstances.size(), 3);
+
+ Set<SchedulableEntityInstance> expectedInputFeedInstances = new HashSet<>();
+ SchedulableEntityInstance instance = new SchedulableEntityInstance(feed.getName(), cluster.getName(),
+ getDate("2012-02-28 00:00 UTC"), EntityType.FEED);
+ instance.setTag(SchedulableEntityInstance.INPUT);
+ expectedInputFeedInstances.add(instance);
+ instance = new SchedulableEntityInstance(feed.getName(), cluster.getName(), getDate("2012-02-28 01:00 UTC"),
+ EntityType.FEED);
+ instance.setTag(SchedulableEntityInstance.INPUT);
+ expectedInputFeedInstances.add(instance);
+ instance = new SchedulableEntityInstance(feed.getName(), cluster.getName(), getDate("2012-02-28 02:00 UTC"),
+ EntityType.FEED);
+ instance.setTag(SchedulableEntityInstance.INPUT);
+ expectedInputFeedInstances.add(instance);
+
+ //Validate with expected result
+ Assert.assertTrue(inputFeedInstances.equals(expectedInputFeedInstances));
+ }
+
+ @Test
+ public void testGetOutputFeedInstances() throws FalconException, ParseException {
+ // create a process with input feeds
+ Cluster cluster = publishCluster();
+ Feed feed = publishFeed(cluster, "days(1)", "2012-02-27 11:00 UTC", "2016-02-28 11:00 UTC");
+ Process process = prepareProcess(cluster, "days(1)", "2012-02-28 10:00 UTC", "2016-02-28 10:00 UTC");
+ Outputs outputs = new Outputs();
+ outputs.getOutputs().add(getOutput("outputFeed", feed.getName(), "now(0,0)"));
+ process.setOutputs(outputs);
+ store.publish(EntityType.PROCESS, process);
+
+ Set<SchedulableEntityInstance> result = ProcessHelper.getOutputFeedInstances(process,
+ getDate("2012-02-28 10:00 UTC"), cluster);
+
+ Set<SchedulableEntityInstance> expected = new HashSet<>();
+ SchedulableEntityInstance ins = new SchedulableEntityInstance(feed.getName(), cluster.getName(),
+ getDate("2012-02-28 11:00 UTC"), EntityType.FEED);
+ ins.setTag(SchedulableEntityInstance.OUTPUT);
+ expected.add(ins);
+
+ Assert.assertEquals(result, expected);
+
+ }
+
+ private org.apache.falcon.entity.v0.process.Validity getProcessValidity(String start, String end) throws
+ ParseException {
+
+ org.apache.falcon.entity.v0.process.Validity validity = new org.apache.falcon.entity.v0.process.Validity();
+ validity.setStart(getDate(start));
+ validity.setEnd(getDate(end));
+ return validity;
+ }
+
+ private Date getDate(String dateString) throws ParseException {
+ return new SimpleDateFormat("yyyy-MM-dd HH:mm Z").parse(dateString);
+ }
+
+ private org.apache.falcon.entity.v0.feed.Validity getFeedValidity(String start, String end) throws ParseException {
+ org.apache.falcon.entity.v0.feed.Validity validity = new org.apache.falcon.entity.v0.feed.Validity();
+ validity.setStart(getDate(start));
+ validity.setEnd(getDate(end));
+ return validity;
+ }
+
+ private Input getInput(String name, String feedName, String start, String end, boolean isOptional) {
+ Input inFeed = new Input();
+ inFeed.setName(name);
+ inFeed.setFeed(feedName);
+ inFeed.setStart(start);
+ inFeed.setEnd(end);
+ inFeed.setOptional(isOptional);
+ return inFeed;
+ }
+
+ private Output getOutput(String name, String feedName, String instance) {
+ Output output = new Output();
+ output.setInstance(instance);
+ output.setFeed(feedName);
+ output.setName(name);
+ return output;
+ }
+
+ private Cluster publishCluster() throws FalconException {
+ Cluster cluster = new Cluster();
+ cluster.setName("feedCluster");
+ cluster.setColo("colo");
+ store.publish(EntityType.CLUSTER, cluster);
+ return cluster;
+
+ }
+
+ private Feed publishFeed(Cluster cluster, String frequency, String start, String end)
+ throws FalconException, ParseException {
+
+ Feed feed = new Feed();
+ feed.setName("feed");
+ Frequency f = new Frequency(frequency);
+ feed.setFrequency(f);
+ feed.setTimezone(UTC);
+ Clusters fClusters = new Clusters();
+ org.apache.falcon.entity.v0.feed.Cluster fCluster = new org.apache.falcon.entity.v0.feed.Cluster();
+ fCluster.setName(cluster.getName());
+ fCluster.setValidity(getFeedValidity(start, end));
+ fClusters.getClusters().add(fCluster);
+ feed.setClusters(fClusters);
+ store.publish(EntityType.FEED, feed);
+
+ return feed;
+ }
+
+ private Process prepareProcess(Cluster cluster, String frequency, String start, String end) throws ParseException {
+ Process process = new Process();
+ process.setName("process");
+ process.setTimezone(UTC);
+ org.apache.falcon.entity.v0.process.Clusters pClusters = new org.apache.falcon.entity.v0.process.Clusters();
+ org.apache.falcon.entity.v0.process.Cluster pCluster = new org.apache.falcon.entity.v0.process.Cluster();
+ pCluster.setName(cluster.getName());
+ org.apache.falcon.entity.v0.process.Validity validity = getProcessValidity(start, end);
+ pCluster.setValidity(validity);
+ pClusters.getClusters().add(pCluster);
+ process.setClusters(pClusters);
+ Frequency f = new Frequency(frequency);
+ process.setFrequency(f);
+ return process;
+ }
+}
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/docs/src/site/twiki/FalconCLI.twiki
----------------------------------------------------------------------
diff --git a/docs/src/site/twiki/FalconCLI.twiki b/docs/src/site/twiki/FalconCLI.twiki
index e447915..50dce84 100644
--- a/docs/src/site/twiki/FalconCLI.twiki
+++ b/docs/src/site/twiki/FalconCLI.twiki
@@ -280,6 +280,41 @@ Usage:
$FALCON_HOME/bin/falcon instance -type <<feed/process>> -name <<name>> -params -start "yyyy-MM-dd'T'HH:mm'Z'"
+
+---+++Dependency
+Display the dependent instances which are dependent on the given instance. For example for a given process instance it will
+list all the input feed instances(if any) and the output feed instances(if any).
+
+An example use case of this command is as follows:
+Suppose you find out that the data in a feed instance was incorrect and you need to figure out which all process instances
+consumed this feed instance so that you can reprocess them after correcting the feed instance. You can give the feed instance
+and it will tell you which process instance produced this feed and which all process instances consumed this feed.
+
+NOTE:
+1. instanceTime must be a valid instanceTime e.g. instanceTime of a feed should be in it's validity range on applicable clusters,
+ and it should be in the range of instances produced by the producer process(if any)
+
+2. For processes with inputs like latest() which vary with time the results are not guaranteed to be correct.
+
+Usage:
+$FALCON_HOME/bin/falcon instance -type <<feed/process>> -name <<name>> -params -instanceTime "yyyy-MM-dd'T'HH:mm'Z'"
+
+For example:
+$FALCON_HOME/bin/falcon instance -dependency -type feed -name out -instanceTime 2014-12-15T00:00Z
+name: producer, type: PROCESS, cluster: local, instanceTime: 2014-12-15T00:00Z, tag: Output
+name: consumer, type: PROCESS, cluster: local, instanceTime: 2014-12-15T00:03Z, tag: Input
+name: consumer, type: PROCESS, cluster: local, instanceTime: 2014-12-15T00:04Z, tag: Input
+name: consumer, type: PROCESS, cluster: local, instanceTime: 2014-12-15T00:02Z, tag: Input
+name: consumer, type: PROCESS, cluster: local, instanceTime: 2014-12-15T00:05Z, tag: Input
+
+
+Response: default/Success!
+
+Request Id: default/1125035965@qtp-503156953-7 - 447be0ad-1d38-4dce-b438-20f3de69b172
+
+
+<a href="./Restapi/InstanceDependency.html">Optional params described here.</a>
+
---++ Metadata Lineage Options
---+++Lineage
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/docs/src/site/twiki/restapi/InstanceDependency.twiki
----------------------------------------------------------------------
diff --git a/docs/src/site/twiki/restapi/InstanceDependency.twiki b/docs/src/site/twiki/restapi/InstanceDependency.twiki
new file mode 100644
index 0000000..dc452de
--- /dev/null
+++ b/docs/src/site/twiki/restapi/InstanceDependency.twiki
@@ -0,0 +1,49 @@
+---++ GET /api/instance/dependency/:entity-type/:entity-name
+ * <a href="#Description">Description</a>
+ * <a href="#Parameters">Parameters</a>
+ * <a href="#Results">Results</a>
+ * <a href="#Examples">Examples</a>
+
+---++ Description
+Get dependent instances for a particular instance.
+
+---++ Parameters
+ * :entity-type Valid options are feed or process.
+ * :entity-name Name of the entity
+ * instanceTime <mandatory param> time of the given instance
+ * colo <optional param> name of the colo
+
+
+---++ Results
+Dependent instances for the specified instance
+
+---++ Examples
+---+++ Rest Call
+<verbatim>
+GET http://localhost:15000/api/instance/dependency/feed/myFeed?colo=*&instanceTime=2012-04-03T07:00Z
+</verbatim>
+---+++ Result
+<verbatim>
+{
+ 'status': 'SUCCEEDED',
+ 'message': 'default/Success!\n',
+ 'dependencies': [
+ {
+ 'cluster': 'local',
+ 'entityName': 'consumer-process',
+ 'entityType': 'PROCESS',
+ 'instanceTime': '2014-12-18T00:00Z',
+ 'tags': 'Input'
+ },
+ {
+ 'cluster': 'local',
+ 'entityName': 'producer-process',
+ 'entityType': 'PROCESS',
+ 'instanceTime': '2014-12-18T00:00Z',
+ 'tags': 'Output'
+ }
+ ],
+ 'requestId': 'default/1405883107@qtp-1501726962-6-0c2e690f-546b-47b0-a5ee-0365d4522a31\n'
+}
+</verbatim>
+
http://git-wip-us.apache.org/repos/asf/falcon/blob/9fd86b78/docs/src/site/twiki/restapi/ResourceList.twiki
----------------------------------------------------------------------
diff --git a/docs/src/site/twiki/restapi/ResourceList.twiki b/docs/src/site/twiki/restapi/ResourceList.twiki
index 060e0af..49dddb7 100644
--- a/docs/src/site/twiki/restapi/ResourceList.twiki
+++ b/docs/src/site/twiki/restapi/ResourceList.twiki
@@ -67,6 +67,7 @@ See also: [[../Security.twiki][Security in Falcon]]
| POST | [[InstanceRerun][api/instance/rerun/:entity-type/:entity-name]] | Rerun a given instance |
| GET | [[InstanceLogs][api/instance/logs/:entity-type/:entity-name]] | Get logs of a given instance |
| GET | [[InstanceSummary][api/instance/summary/:entity-type/:entity-name]] | Return summary of instances for an entity |
+| GET | [[InstanceDependency][api/instance/dependency/:entity-type/:entity-name]] | Return dependent instances for a given instance |
---++ REST Call on Metadata Lineage Resource