You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@falcon.apache.org by sr...@apache.org on 2013/04/26 17:50:29 UTC
[13/47] Fixes for Checkstyle
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java b/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java
index 14d35aa..1f3f256 100644
--- a/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java
+++ b/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java
@@ -18,15 +18,15 @@
package org.apache.falcon.aspect;
-import java.util.HashMap;
-import java.util.Map;
-
import org.apache.falcon.util.ResourcesReflectionUtil;
import org.apache.log4j.Logger;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
+import java.util.HashMap;
+import java.util.Map;
+
/**
* Abstract Falcon Aspect, which intercept methods annotated with Monitored and
* publishes messages. Subclasses should override publishMessage Method.
@@ -34,65 +34,65 @@ import org.aspectj.lang.annotation.Aspect;
@Aspect
public abstract class AbstractFalconAspect {
- private static final Logger LOG = Logger
- .getLogger(AbstractFalconAspect.class);
+ private static final Logger LOG = Logger
+ .getLogger(AbstractFalconAspect.class);
- @Around("@annotation(org.apache.falcon.monitors.Monitored)")
- public Object logAround(ProceedingJoinPoint joinPoint) throws Throwable {
+ @Around("@annotation(org.apache.falcon.monitors.Monitored)")
+ public Object logAround(ProceedingJoinPoint joinPoint) throws Throwable {
- String methodName = joinPoint.getSignature().getName();
- Object[] args = joinPoint.getArgs();
- Object result = null;
- ResourceMessage.Status status;
+ String methodName = joinPoint.getSignature().getName();
+ Object[] args = joinPoint.getArgs();
+ Object result = null;
+ ResourceMessage.Status status;
- long startTime = System.nanoTime();
- long endTime;
- try {
- result = joinPoint.proceed();
- } catch (Exception e) {
- endTime = System.nanoTime();
- status = ResourceMessage.Status.FAILED;
- publishMessage(getResourceMessage(joinPoint.getSignature()
- .getDeclaringType().getSimpleName()
- + "." + methodName, args, status, endTime - startTime));
- throw e;
- }
- endTime = System.nanoTime();
- status = ResourceMessage.Status.SUCCEEDED;
- publishMessage(getResourceMessage(joinPoint.getSignature()
- .getDeclaringType().getSimpleName()
- + "." + methodName, args, status, endTime - startTime));
- return result;
- }
+ long startTime = System.nanoTime();
+ long endTime;
+ try {
+ result = joinPoint.proceed();
+ } catch (Exception e) {
+ endTime = System.nanoTime();
+ status = ResourceMessage.Status.FAILED;
+ publishMessage(getResourceMessage(joinPoint.getSignature()
+ .getDeclaringType().getSimpleName()
+ + "." + methodName, args, status, endTime - startTime));
+ throw e;
+ }
+ endTime = System.nanoTime();
+ status = ResourceMessage.Status.SUCCEEDED;
+ publishMessage(getResourceMessage(joinPoint.getSignature()
+ .getDeclaringType().getSimpleName()
+ + "." + methodName, args, status, endTime - startTime));
+ return result;
+ }
- private ResourceMessage getResourceMessage(String methodName,
- Object[] args, ResourceMessage.Status status, long executionTime) {
- String action = ResourcesReflectionUtil
- .getResourceMonitorName(methodName);
+ private ResourceMessage getResourceMessage(String methodName,
+ Object[] args, ResourceMessage.Status status, long executionTime) {
+ String action = ResourcesReflectionUtil
+ .getResourceMonitorName(methodName);
- assert action != null : "Method :" + methodName
- + " not parsed by reflection util";
- Map<String, String> dimensions = new HashMap<String, String>();
+ assert action != null : "Method :" + methodName
+ + " not parsed by reflection util";
+ Map<String, String> dimensions = new HashMap<String, String>();
- if (ResourcesReflectionUtil.getResourceDimensionsName(methodName) == null) {
- LOG.warn("Class for method name: " + methodName
- + " is not added to ResourcesReflectionUtil");
- } else {
- for (Map.Entry<Integer, String> param : ResourcesReflectionUtil
- .getResourceDimensionsName(methodName).entrySet()) {
- dimensions.put(
- param.getValue(),
- args[param.getKey()] == null ? "NULL" : args[param
- .getKey()].toString());
- }
- }
- Integer timeTakenArg = ResourcesReflectionUtil
- .getResourceTimeTakenName(methodName);
- return timeTakenArg == null ? new ResourceMessage(action, dimensions,
- status, executionTime) : new ResourceMessage(action,
- dimensions, status, Long.valueOf(args[timeTakenArg]
- .toString()));
- }
+ if (ResourcesReflectionUtil.getResourceDimensionsName(methodName) == null) {
+ LOG.warn("Class for method name: " + methodName
+ + " is not added to ResourcesReflectionUtil");
+ } else {
+ for (Map.Entry<Integer, String> param : ResourcesReflectionUtil
+ .getResourceDimensionsName(methodName).entrySet()) {
+ dimensions.put(
+ param.getValue(),
+ args[param.getKey()] == null ? "NULL" : args[param
+ .getKey()].toString());
+ }
+ }
+ Integer timeTakenArg = ResourcesReflectionUtil
+ .getResourceTimeTakenName(methodName);
+ return timeTakenArg == null ? new ResourceMessage(action, dimensions,
+ status, executionTime) : new ResourceMessage(action,
+ dimensions, status, Long.valueOf(args[timeTakenArg]
+ .toString()));
+ }
- abstract public void publishMessage(ResourceMessage message);
+ abstract public void publishMessage(ResourceMessage message);
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java b/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java
index 9705f9e..b792434 100644
--- a/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java
+++ b/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java
@@ -29,83 +29,83 @@ import org.aspectj.lang.annotation.Aspect;
@Aspect
public class GenericAlert {
- @Monitored(event = "retry-instance-failed")
- public static String alertRetryFailed(
- @Dimension(value = "entity-type") String entityType,
- @Dimension(value = "entity-name") String entityName,
- @Dimension(value = "nominal-name") String nominalTime,
- @Dimension(value = "wf-id") String wfId,
- @Dimension(value = "run-id") String runId,
- @Dimension(value = "error-message") String message) {
- return "IGNORE";
- }
-
- @Monitored(event = "late-rerun-failed")
- public static String alertLateRerunFailed(
- @Dimension(value = "entity-type") String entityType,
- @Dimension(value = "entity-name") String entityName,
- @Dimension(value = "nominal-name") String nominalTime,
- @Dimension(value = "wf-id") String wfId,
- @Dimension(value = "run-id") String runId,
- @Dimension(value = "error-message") String message) {
- return "IGNORE";
+ @Monitored(event = "retry-instance-failed")
+ public static String alertRetryFailed(
+ @Dimension(value = "entity-type") String entityType,
+ @Dimension(value = "entity-name") String entityName,
+ @Dimension(value = "nominal-name") String nominalTime,
+ @Dimension(value = "wf-id") String wfId,
+ @Dimension(value = "run-id") String runId,
+ @Dimension(value = "error-message") String message) {
+ return "IGNORE";
+ }
+
+ @Monitored(event = "late-rerun-failed")
+ public static String alertLateRerunFailed(
+ @Dimension(value = "entity-type") String entityType,
+ @Dimension(value = "entity-name") String entityName,
+ @Dimension(value = "nominal-name") String nominalTime,
+ @Dimension(value = "wf-id") String wfId,
+ @Dimension(value = "run-id") String runId,
+ @Dimension(value = "error-message") String message) {
+ return "IGNORE";
- }
+ }
- @Monitored(event = "wf-instance-failed")
- public static String instrumentFailedInstance(
- @Dimension(value = "cluster") String cluster,
- @Dimension(value = "entity-type") String entityType,
- @Dimension(value = "entity-name") String entityName,
- @Dimension(value = "nominal-time") String nominalTime,
- @Dimension(value = "wf-id") String workflowId,
- @Dimension(value = "run-id") String runId,
- @Dimension(value = "operation") String operation,
+ @Monitored(event = "wf-instance-failed")
+ public static String instrumentFailedInstance(
+ @Dimension(value = "cluster") String cluster,
+ @Dimension(value = "entity-type") String entityType,
+ @Dimension(value = "entity-name") String entityName,
+ @Dimension(value = "nominal-time") String nominalTime,
+ @Dimension(value = "wf-id") String workflowId,
+ @Dimension(value = "run-id") String runId,
+ @Dimension(value = "operation") String operation,
@Dimension(value = "start-time") String startTime,
@Dimension(value = "error-message") String errorMessage,
@Dimension(value = "message") String message,
- @TimeTaken long timeTaken)
- throws Exception {
- return "IGNORE";
- }
-
- @Monitored(event = "wf-instance-succeeded")
- public static String instrumentSucceededInstance(
- @Dimension(value = "cluster") String cluster,
- @Dimension(value = "entity-type") String entityType,
- @Dimension(value = "entity-name") String entityName,
- @Dimension(value = "nominal-time") String nominalTime,
- @Dimension(value = "wf-id") String workflowId,
- @Dimension(value = "run-id") String runId,
- @Dimension(value = "operation") String operation,
+ @TimeTaken long timeTaken)
+ throws Exception {
+ return "IGNORE";
+ }
+
+ @Monitored(event = "wf-instance-succeeded")
+ public static String instrumentSucceededInstance(
+ @Dimension(value = "cluster") String cluster,
+ @Dimension(value = "entity-type") String entityType,
+ @Dimension(value = "entity-name") String entityName,
+ @Dimension(value = "nominal-time") String nominalTime,
+ @Dimension(value = "wf-id") String workflowId,
+ @Dimension(value = "run-id") String runId,
+ @Dimension(value = "operation") String operation,
@Dimension(value = "start-time") String startTime,
- @TimeTaken long timeTaken)
- throws Exception {
- return "IGNORE";
- }
-
- @Monitored(event = "rerun-queue-failed")
- public static String alertRerunConsumerFailed(
- @Dimension(value = "message") String message,
- @Dimension(value = "exception") Exception exception) {
- return "IGNORE";
+ @TimeTaken long timeTaken)
+ throws Exception {
+ return "IGNORE";
+ }
- }
+ @Monitored(event = "rerun-queue-failed")
+ public static String alertRerunConsumerFailed(
+ @Dimension(value = "message") String message,
+ @Dimension(value = "exception") Exception exception) {
+ return "IGNORE";
+
+ }
- @Monitored(event = "sla-miss")
- public static String alertOnLikelySLAMiss(
+ @Monitored(event = "sla-miss")
+ public static String alertOnLikelySLAMiss(
@Dimension(value = "cluster") String cluster,
@Dimension(value = "entity-type") String entityType,
@Dimension(value = "entity-name") String entityName,
@Dimension(value = "nominal-time") String nominalTime) {
- return "IGNORE";
- }
-
- @Monitored(event = "log-cleanup-service-failed")
- public static String alertLogCleanupServiceFailed(
- @Dimension(value = "message") String message,
- @Dimension(value = "exception") Throwable throwable) {
- return "IGNORE";
+ return "IGNORE";
+ }
+
+ @Monitored(event = "log-cleanup-service-failed")
+ public static String alertLogCleanupServiceFailed(
+ @Dimension(value = "message") String message,
+ @Dimension(value = "exception") Throwable throwable) {
+ return "IGNORE";
- }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/metrics/src/main/java/org/apache/falcon/aspect/ResourceMessage.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/aspect/ResourceMessage.java b/metrics/src/main/java/org/apache/falcon/aspect/ResourceMessage.java
index b00cbca..c7ad8e7 100644
--- a/metrics/src/main/java/org/apache/falcon/aspect/ResourceMessage.java
+++ b/metrics/src/main/java/org/apache/falcon/aspect/ResourceMessage.java
@@ -23,43 +23,44 @@ import java.util.Map;
//Message to be sent to logging system
public class ResourceMessage {
- private String action;
- private Map<String, String> dimensions;
- private Status status;
- private long executionTime;
-
- public enum Status{
- SUCCEEDED, FAILED
- }
+ private String action;
+ private Map<String, String> dimensions;
+ private Status status;
+ private long executionTime;
- public ResourceMessage(String action, Map<String, String> dimensions,
- Status status, long executionTime) {
- this.action = action;
- this.dimensions = dimensions;
- this.status = status;
- this.executionTime = executionTime;
- }
-
- public String getAction() {
- return action;
- }
+ public enum Status {
+ SUCCEEDED, FAILED
+ }
- public Map<String, String> getDimensions() {
- return dimensions;
- }
+ public ResourceMessage(String action, Map<String, String> dimensions,
+ Status status, long executionTime) {
+ this.action = action;
+ this.dimensions = dimensions;
+ this.status = status;
+ this.executionTime = executionTime;
+ }
- public Status getStatus() {
- return status;
- }
+ public String getAction() {
+ return action;
+ }
- public long getExecutionTime() {
- return executionTime;
- }
+ public Map<String, String> getDimensions() {
+ return dimensions;
+ }
- @Override
- public String toString() {
- return "{Action:"+action+", Dimensions:"+dimensions+", Status: "+status.name()+", Time-taken:"+executionTime+" ns}";
- }
+ public Status getStatus() {
+ return status;
+ }
+
+ public long getExecutionTime() {
+ return executionTime;
+ }
+
+ @Override
+ public String toString() {
+ return "{Action:" + action + ", Dimensions:" + dimensions + ", Status: " + status.name() + ", Time-taken:"
+ + executionTime + " ns}";
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java b/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java
index 2f8dd6d..119a7f7 100644
--- a/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java
+++ b/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java
@@ -21,7 +21,7 @@ package org.apache.falcon.util;
import org.apache.falcon.monitors.Monitored;
import org.apache.falcon.monitors.TimeTaken;
-import java.awt.Dimension;
+import java.awt.*;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Collections;
@@ -31,55 +31,54 @@ import java.util.Map;
/**
* Builds a cached of methods annotated with Monitored and params of methods
* annotated with Dimension.
- *
*/
public class ResourcesReflectionUtil {
- private static final Map<String, MethodAnnotation> methods = new HashMap<String, MethodAnnotation>();
-
- private ResourcesReflectionUtil(){
-
- }
-
- static {
- //TODO load these classes from properties file
- buildAnnotationsMapForClass("org.apache.falcon.resource.proxy.SchedulableEntityManagerProxy");
- buildAnnotationsMapForClass("org.apache.falcon.resource.proxy.InstanceManagerProxy");
- buildAnnotationsMapForClass("org.apache.falcon.resource.AbstractInstanceManager");
- buildAnnotationsMapForClass("org.apache.falcon.service.FalconTopicSubscriber");
- buildAnnotationsMapForClass("org.apache.falcon.aspect.GenericAlert");
- }
-
- public static Map<Integer, String> getResourceDimensionsName(String methodName) {
- return methods.get(methodName)!=null?Collections.unmodifiableMap(methods.get(methodName).params):null;
- }
-
- public static String getResourceMonitorName(String methodName) {
- return methods.get(methodName)!=null?methods.get(methodName).monitoredName:null;
- }
-
- public static Integer getResourceTimeTakenName(String methodName) {
- return methods.get(methodName) != null ? methods.get(methodName).timeTakenArgIndex
- : null;
- }
-
- public static class MethodAnnotation {
- private String monitoredName;
- // argument <index,DimensionValue>
- private Map<Integer, String> params = new HashMap<Integer, String>();
-
- //to override time taken by an api
- private Integer timeTakenArgIndex;
-
- @Override
- public String toString() {
- return "{" + monitoredName + "[" + params.toString() + "]" + "}";
- }
-
- }
-
- private static void buildAnnotationsMapForClass(String className) {
- Class clazz;
+ private static final Map<String, MethodAnnotation> methods = new HashMap<String, MethodAnnotation>();
+
+ private ResourcesReflectionUtil() {
+
+ }
+
+ static {
+ //TODO load these classes from properties file
+ buildAnnotationsMapForClass("org.apache.falcon.resource.proxy.SchedulableEntityManagerProxy");
+ buildAnnotationsMapForClass("org.apache.falcon.resource.proxy.InstanceManagerProxy");
+ buildAnnotationsMapForClass("org.apache.falcon.resource.AbstractInstanceManager");
+ buildAnnotationsMapForClass("org.apache.falcon.service.FalconTopicSubscriber");
+ buildAnnotationsMapForClass("org.apache.falcon.aspect.GenericAlert");
+ }
+
+ public static Map<Integer, String> getResourceDimensionsName(String methodName) {
+ return methods.get(methodName) != null ? Collections.unmodifiableMap(methods.get(methodName).params) : null;
+ }
+
+ public static String getResourceMonitorName(String methodName) {
+ return methods.get(methodName) != null ? methods.get(methodName).monitoredName : null;
+ }
+
+ public static Integer getResourceTimeTakenName(String methodName) {
+ return methods.get(methodName) != null ? methods.get(methodName).timeTakenArgIndex
+ : null;
+ }
+
+ public static class MethodAnnotation {
+ private String monitoredName;
+ // argument <index,DimensionValue>
+ private Map<Integer, String> params = new HashMap<Integer, String>();
+
+ //to override time taken by an api
+ private Integer timeTakenArgIndex;
+
+ @Override
+ public String toString() {
+ return "{" + monitoredName + "[" + params.toString() + "]" + "}";
+ }
+
+ }
+
+ private static void buildAnnotationsMapForClass(String className) {
+ Class clazz;
try {
clazz = ResourcesReflectionUtil.class.
getClassLoader().loadClass(className);
@@ -88,7 +87,7 @@ public class ResourcesReflectionUtil {
}
Method[] declMethods = clazz.getMethods();
- // scan every method
+ // scan every method
for (Method declMethod : declMethods) {
Annotation[] methodAnnots = declMethod.getDeclaredAnnotations();
// scan every annotation on method
@@ -109,41 +108,41 @@ public class ResourcesReflectionUtil {
}
}
- }
-
- private static Map<Integer, String> getDeclaredParamAnnots(
- Annotation[][] paramAnnots, MethodAnnotation annotation) {
- Map<Integer, String> params = new HashMap<Integer, String>();
- for (int i = 0; i < paramAnnots.length; i++) {
- for (int j = 0; j < paramAnnots[i].length; j++) {
- if (paramAnnots[i][j].annotationType().getSimpleName()
- .equals(Dimension.class.getSimpleName())) {
- params.put(i, getAnnotationValue(paramAnnots[i][j], "value"));
- }
- if (paramAnnots[i][j].annotationType().getSimpleName()
- .equals(TimeTaken.class.getSimpleName())) {
- annotation.timeTakenArgIndex = i;
- }
- }
- }
- return params;
-
- }
-
- private static String getAnnotationValue(Annotation annotation,
- String attributeName) {
- String value = null;
-
- if (annotation != null) {
- try {
- value = (String) annotation.annotationType()
- .getMethod(attributeName).invoke(annotation);
- } catch (Exception ignore) {
- ignore.printStackTrace();
- }
- }
-
- return value;
- }
+ }
+
+ private static Map<Integer, String> getDeclaredParamAnnots(
+ Annotation[][] paramAnnots, MethodAnnotation annotation) {
+ Map<Integer, String> params = new HashMap<Integer, String>();
+ for (int i = 0; i < paramAnnots.length; i++) {
+ for (int j = 0; j < paramAnnots[i].length; j++) {
+ if (paramAnnots[i][j].annotationType().getSimpleName()
+ .equals(Dimension.class.getSimpleName())) {
+ params.put(i, getAnnotationValue(paramAnnots[i][j], "value"));
+ }
+ if (paramAnnots[i][j].annotationType().getSimpleName()
+ .equals(TimeTaken.class.getSimpleName())) {
+ annotation.timeTakenArgIndex = i;
+ }
+ }
+ }
+ return params;
+
+ }
+
+ private static String getAnnotationValue(Annotation annotation,
+ String attributeName) {
+ String value = null;
+
+ if (annotation != null) {
+ try {
+ value = (String) annotation.annotationType()
+ .getMethod(attributeName).invoke(annotation);
+ } catch (Exception ignore) {
+ ignore.printStackTrace();
+ }
+ }
+
+ return value;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java b/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
index 915a69c..ac3e76e 100644
--- a/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
+++ b/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
@@ -18,24 +18,7 @@
package org.apache.falcon.converter;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.io.StringWriter;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import javax.xml.bind.JAXBContext;
-import javax.xml.bind.JAXBElement;
-import javax.xml.bind.JAXBException;
-import javax.xml.bind.Marshaller;
-import javax.xml.bind.Unmarshaller;
-
import org.apache.commons.lang.StringUtils;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.falcon.FalconException;
import org.apache.falcon.FalconRuntimException;
import org.apache.falcon.Tag;
@@ -54,9 +37,21 @@ import org.apache.falcon.oozie.workflow.WORKFLOWAPP;
import org.apache.falcon.service.FalconPathFilter;
import org.apache.falcon.service.SharedLibraryHostingService;
import org.apache.falcon.util.StartupProperties;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.log4j.Logger;
import org.apache.oozie.client.OozieClient;
+import javax.xml.bind.*;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.StringWriter;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
public abstract class AbstractOozieEntityMapper<T extends Entity> {
private static Logger LOG = Logger.getLogger(AbstractOozieEntityMapper.class);
@@ -65,8 +60,8 @@ public abstract class AbstractOozieEntityMapper<T extends Entity> {
protected static final String ACTUAL_TIME_EL = "${coord:formatTime(coord:actualTime(), 'yyyy-MM-dd-HH-mm')}";
protected static final Long DEFAULT_BROKER_MSG_TTL = 3 * 24 * 60L;
- protected static final String MR_QUEUE_NAME="queueName";
- protected static final String MR_JOB_PRIORITY="jobPriority";
+ protected static final String MR_QUEUE_NAME = "queueName";
+ protected static final String MR_JOB_PRIORITY = "jobPriority";
protected static final JAXBContext workflowJaxbContext;
protected static final JAXBContext coordJaxbContext;
@@ -75,16 +70,18 @@ public abstract class AbstractOozieEntityMapper<T extends Entity> {
protected static final FalconPathFilter falconJarFilter = new FalconPathFilter() {
@Override
public boolean accept(Path path) {
- if (path.getName().startsWith("falcon"))
+ if (path.getName().startsWith("falcon")) {
return true;
+ }
return false;
}
@Override
public String getJarName(Path path) {
String name = path.getName();
- if(name.endsWith(".jar"))
+ if (name.endsWith(".jar")) {
name = name.substring(0, name.indexOf(".jar"));
+ }
return name;
}
};
@@ -127,7 +124,8 @@ public abstract class AbstractOozieEntityMapper<T extends Entity> {
}
for (COORDINATORAPP coordinatorapp : coordinators) {
Path coordPath = getCoordPath(bundlePath, coordinatorapp.getName());
- String coordXmlName = marshal(cluster, coordinatorapp, coordPath, EntityUtil.getWorkflowNameSuffix(coordinatorapp.getName(), entity));
+ String coordXmlName = marshal(cluster, coordinatorapp, coordPath,
+ EntityUtil.getWorkflowNameSuffix(coordinatorapp.getName(), entity));
createTempDir(cluster, coordPath);
COORDINATOR bundleCoord = new COORDINATOR();
bundleCoord.setName(coordinatorapp.getName());
@@ -145,23 +143,26 @@ public abstract class AbstractOozieEntityMapper<T extends Entity> {
try {
Path libPath = new Path(coordPath, "lib");
FileSystem fs = FileSystem.get(ClusterHelper.getConfiguration(cluster));
- if (!fs.exists(libPath))
+ if (!fs.exists(libPath)) {
fs.mkdirs(libPath);
+ }
SharedLibraryHostingService.pushLibsToHDFS(libPath.toString(), cluster, falconJarFilter);
} catch (IOException e) {
LOG.error("Failed to copy shared libs on cluster " + cluster.getName(), e);
- throw new FalconException("Failed to copy shared libs on cluster " + cluster.getName(),e);
+ throw new FalconException("Failed to copy shared libs on cluster " + cluster.getName(), e);
}
}
protected abstract List<COORDINATORAPP> getCoordinators(Cluster cluster, Path bundlePath) throws FalconException;
protected org.apache.falcon.oozie.coordinator.CONFIGURATION getCoordConfig(Map<String, String> propMap) {
- org.apache.falcon.oozie.coordinator.CONFIGURATION conf = new org.apache.falcon.oozie.coordinator.CONFIGURATION();
+ org.apache.falcon.oozie.coordinator.CONFIGURATION conf
+ = new org.apache.falcon.oozie.coordinator.CONFIGURATION();
List<org.apache.falcon.oozie.coordinator.CONFIGURATION.Property> props = conf.getProperty();
- for (Entry<String, String> prop : propMap.entrySet())
+ for (Entry<String, String> prop : propMap.entrySet()) {
props.add(createCoordProperty(prop.getKey(), prop.getValue()));
+ }
return conf;
}
@@ -172,37 +173,42 @@ public abstract class AbstractOozieEntityMapper<T extends Entity> {
props.put(ARG.timeStamp.getPropName(), ACTUAL_TIME_EL);
props.put("userBrokerUrl", ClusterHelper.getMessageBrokerUrl(cluster));
props.put("userBrokerImplClass", ClusterHelper.getMessageBrokerImplClass(cluster));
- String falconBrokerUrl = StartupProperties.get().getProperty(ARG.brokerUrl.getPropName(), "tcp://localhost:61616?daemon=true");
+ String falconBrokerUrl = StartupProperties.get().getProperty(ARG.brokerUrl.getPropName(),
+ "tcp://localhost:61616?daemon=true");
props.put(ARG.brokerUrl.getPropName(), falconBrokerUrl);
String falconBrokerImplClass = StartupProperties.get().getProperty(ARG.brokerImplClass.getPropName(),
ClusterHelper.DEFAULT_BROKER_IMPL_CLASS);
props.put(ARG.brokerImplClass.getPropName(), falconBrokerImplClass);
- String jmsMessageTTL = StartupProperties.get().getProperty("broker.ttlInMins", DEFAULT_BROKER_MSG_TTL.toString());
+ String jmsMessageTTL = StartupProperties.get().getProperty("broker.ttlInMins",
+ DEFAULT_BROKER_MSG_TTL.toString());
props.put(ARG.brokerTTL.getPropName(), jmsMessageTTL);
props.put(ARG.entityType.getPropName(), entity.getEntityType().name());
props.put("logDir", getStoragePath(new Path(coordPath, "../../logs")));
- props.put(OozieClient.EXTERNAL_ID, new ExternalId(entity.getName(), EntityUtil.getWorkflowNameTag(coordName, entity),
- "${coord:nominalTime()}").getId());
+ props.put(OozieClient.EXTERNAL_ID,
+ new ExternalId(entity.getName(), EntityUtil.getWorkflowNameTag(coordName, entity),
+ "${coord:nominalTime()}").getId());
props.put("workflowEngineUrl", ClusterHelper.getOozieUrl(cluster));
- try {
- if (EntityUtil.getLateProcess(entity) == null
- || EntityUtil.getLateProcess(entity).getLateInputs() == null
- || EntityUtil.getLateProcess(entity).getLateInputs().size() == 0) {
- props.put("shouldRecord", "false");
- } else {
- props.put("shouldRecord", "true");
- }
- } catch (FalconException e) {
- LOG.error("Unable to get Late Process for entity:" + entity, e);
- throw new FalconRuntimException(e);
- }
+ try {
+ if (EntityUtil.getLateProcess(entity) == null
+ || EntityUtil.getLateProcess(entity).getLateInputs() == null
+ || EntityUtil.getLateProcess(entity).getLateInputs().size() == 0) {
+ props.put("shouldRecord", "false");
+ } else {
+ props.put("shouldRecord", "true");
+ }
+ } catch (FalconException e) {
+ LOG.error("Unable to get Late Process for entity:" + entity, e);
+ throw new FalconRuntimException(e);
+ }
props.put("entityName", entity.getName());
props.put("entityType", entity.getEntityType().name().toLowerCase());
props.put(ARG.cluster.getPropName(), cluster.getName());
- if(cluster.getProperties() != null)
- for(Property prop:cluster.getProperties().getProperties())
+ if (cluster.getProperties() != null) {
+ for (Property prop : cluster.getProperties().getProperties()) {
props.put(prop.getName(), prop.getValue());
-
+ }
+ }
+
props.put(MR_QUEUE_NAME, "default");
props.put(MR_JOB_PRIORITY, "NORMAL");
//props in entity override the set props.
@@ -210,21 +216,25 @@ public abstract class AbstractOozieEntityMapper<T extends Entity> {
return props;
}
- protected org.apache.falcon.oozie.coordinator.CONFIGURATION.Property createCoordProperty(String name, String value) {
- org.apache.falcon.oozie.coordinator.CONFIGURATION.Property prop = new org.apache.falcon.oozie.coordinator.CONFIGURATION.Property();
+ protected org.apache.falcon.oozie.coordinator.CONFIGURATION.Property createCoordProperty(String name,
+ String value) {
+ org.apache.falcon.oozie.coordinator.CONFIGURATION.Property prop
+ = new org.apache.falcon.oozie.coordinator.CONFIGURATION.Property();
prop.setName(name);
prop.setValue(value);
return prop;
}
protected org.apache.falcon.oozie.bundle.CONFIGURATION.Property createBundleProperty(String name, String value) {
- org.apache.falcon.oozie.bundle.CONFIGURATION.Property prop = new org.apache.falcon.oozie.bundle.CONFIGURATION.Property();
+ org.apache.falcon.oozie.bundle.CONFIGURATION.Property prop
+ = new org.apache.falcon.oozie.bundle.CONFIGURATION.Property();
prop.setName(name);
prop.setValue(value);
return prop;
}
- protected void marshal(Cluster cluster, JAXBElement<?> jaxbElement, JAXBContext jaxbContext, Path outPath) throws FalconException {
+ protected void marshal(Cluster cluster, JAXBElement<?> jaxbElement, JAXBContext jaxbContext, Path outPath)
+ throws FalconException {
try {
Marshaller marshaller = jaxbContext.createMarshaller();
marshaller.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, Boolean.TRUE);
@@ -260,8 +270,9 @@ public abstract class AbstractOozieEntityMapper<T extends Entity> {
}
protected String marshal(Cluster cluster, COORDINATORAPP coord, Path outPath, String name) throws FalconException {
- if(StringUtils.isEmpty(name))
+ if (StringUtils.isEmpty(name)) {
name = "coordinator";
+ }
name = name + ".xml";
marshal(cluster, new ObjectFactory().createCoordinatorApp(coord), coordJaxbContext, new Path(outPath, name));
return name;
@@ -269,26 +280,30 @@ public abstract class AbstractOozieEntityMapper<T extends Entity> {
protected void marshal(Cluster cluster, BUNDLEAPP bundle, Path outPath) throws FalconException {
- marshal(cluster, new org.apache.falcon.oozie.bundle.ObjectFactory().createBundleApp(bundle), bundleJaxbContext, new Path(
- outPath, "bundle.xml"));
+ marshal(cluster, new org.apache.falcon.oozie.bundle.ObjectFactory().createBundleApp(bundle), bundleJaxbContext,
+ new Path(
+ outPath, "bundle.xml"));
}
protected void marshal(Cluster cluster, WORKFLOWAPP workflow, Path outPath) throws FalconException {
- marshal(cluster, new org.apache.falcon.oozie.workflow.ObjectFactory().createWorkflowApp(workflow), workflowJaxbContext,
+ marshal(cluster, new org.apache.falcon.oozie.workflow.ObjectFactory().createWorkflowApp(workflow),
+ workflowJaxbContext,
new Path(outPath, "workflow.xml"));
}
protected String getStoragePath(Path path) {
- if (path != null)
+ if (path != null) {
return getStoragePath(path.toString());
+ }
return null;
}
protected String getStoragePath(String path) {
if (StringUtils.isNotEmpty(path)) {
- if (new Path(path).toUri().getScheme()==null)
+ if (new Path(path).toUri().getScheme() == null) {
path = "${nameNode}" + path;
+ }
}
return path;
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/oozie/src/main/java/org/apache/falcon/logging/LogMover.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/logging/LogMover.java b/oozie/src/main/java/org/apache/falcon/logging/LogMover.java
index 4d28ded..261fc53 100644
--- a/oozie/src/main/java/org/apache/falcon/logging/LogMover.java
+++ b/oozie/src/main/java/org/apache/falcon/logging/LogMover.java
@@ -17,190 +17,182 @@
*/
package org.apache.falcon.logging;
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.URL;
-import java.net.URLConnection;
-import java.util.List;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.GnuParser;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.*;
+import org.apache.falcon.entity.v0.EntityType;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.mapred.JobClient;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.JobID;
-import org.apache.hadoop.mapred.RunningJob;
-import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
-import org.apache.falcon.entity.v0.EntityType;
import org.apache.log4j.Logger;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.OozieClientException;
import org.apache.oozie.client.WorkflowAction;
import org.apache.oozie.client.WorkflowJob;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.URL;
+import java.net.URLConnection;
+import java.util.List;
+
public class LogMover extends Configured implements Tool {
- private static final Logger LOG = Logger.getLogger(LogMover.class);
-
- private static class ARGS {
- String oozieUrl;
- String subflowId;
- String runId;
- String logDir;
- String status;
- String entityType;
- }
-
- public static void main(String[] args) throws Exception {
- ToolRunner.run(new LogMover(), args);
- }
-
- @Override
- public int run(String[] arguments) throws Exception {
- try {
- ARGS args = new ARGS();
- setupArgs(arguments, args);
- OozieClient client = new OozieClient(args.oozieUrl);
- WorkflowJob jobInfo = null;
- try {
- jobInfo = client.getJobInfo(args.subflowId);
- } catch (OozieClientException e) {
- LOG.error("Error getting jobinfo for: " + args.subflowId, e);
- return 0;
- }
- Path path = new Path(args.logDir + "/"
- + String.format("%03d", Integer.parseInt(args.runId)));
-
- FileSystem fs = path.getFileSystem(getConf());
-
- if (args.entityType.equalsIgnoreCase(EntityType.FEED.name())) {
- // if replication wf
- copyOozieLog(client, fs, path, jobInfo.getId());
- copyTTlogs(args, fs, path, jobInfo.getActions().get(2));
- } else {
- // if process wf
- String subflowId = jobInfo.getExternalId();
- copyOozieLog(client, fs, path, subflowId);
- WorkflowJob subflowInfo = client.getJobInfo(subflowId);
- List<WorkflowAction> actions = subflowInfo.getActions();
- for (WorkflowAction action : actions) {
- if (action.getType().equals("pig")
- || action.getType().equals("java")) {
- copyTTlogs(args, fs, path, action);
- } else {
- LOG.info("Ignoring hadoop TT log for non-pig and non-java action:"
- + action.getName());
- }
- }
-
- }
-
- } catch (Exception e) {
- LOG.error("Exception in log mover:", e);
- }
- return 0;
- }
-
- private void copyOozieLog(OozieClient client, FileSystem fs, Path path,
- String id) throws OozieClientException, IOException {
- InputStream in = new ByteArrayInputStream(client.getJobLog(id).getBytes());
- OutputStream out = fs.create(new Path(path, "oozie.log"));
- IOUtils.copyBytes(in, out, 4096, true);
- LOG.info("Copied oozie log to " + path);
- }
-
- private void copyTTlogs(ARGS args, FileSystem fs, Path path,
- WorkflowAction action) throws Exception {
- String ttLogURL = getTTlogURL(action.getExternalId());
- if (ttLogURL != null) {
- LOG.info("Fetching log for action: " + action.getExternalId()
- + " from url: " + ttLogURL);
- InputStream in = getURLinputStream(new URL(ttLogURL));
- OutputStream out = fs.create(new Path(path, action.getName() + "_"
- + getMappedStatus(action.getStatus()) + ".log"));
- IOUtils.copyBytes(in, out, 4096, true);
- LOG.info("Copied log to " + path);
- }
- }
-
- private String getMappedStatus(WorkflowAction.Status status) {
- if (status == WorkflowAction.Status.FAILED
- || status == WorkflowAction.Status.KILLED
- || status == WorkflowAction.Status.ERROR) {
- return "FAILED";
- } else {
- return "SUCCEEDED";
- }
- }
-
- private void setupArgs(String[] arguments, ARGS args) throws ParseException {
- Options options = new Options();
- Option opt;
- opt = new Option("workflowEngineUrl", true,
- "url of workflow engine, ex:oozie");
- opt.setRequired(true);
- options.addOption(opt);
- opt = new Option("subflowId", true, "external id of userworkflow");
- opt.setRequired(true);
- options.addOption(opt);
- opt = new Option("runId", true, "current workflow's runid");
- opt.setRequired(true);
- options.addOption(opt);
- opt = new Option("logDir", true, "log dir where job logs are stored");
- opt.setRequired(true);
- options.addOption(opt);
- opt = new Option("status", true, "user workflow status");
- opt.setRequired(true);
- options.addOption(opt);
- opt = new Option("entityType", true, "entity type feed or process");
- opt.setRequired(true);
- options.addOption(opt);
-
- CommandLine cmd = new GnuParser().parse(options, arguments);
-
- args.oozieUrl = cmd.getOptionValue("workflowEngineUrl");
- args.subflowId = cmd.getOptionValue("subflowId");
- args.runId = cmd.getOptionValue("runId");
- args.logDir = cmd.getOptionValue("logDir");
- args.status = cmd.getOptionValue("status");
- args.entityType = cmd.getOptionValue("entityType");
-
- }
-
- private String getTTlogURL(String jobId) throws Exception {
- JobConf jobConf = new JobConf(getConf());
- JobClient jobClient = new JobClient(jobConf);
- RunningJob job = jobClient.getJob(JobID.forName(jobId));
- if (job == null) {
- LOG.warn("No running job for job id: " + jobId);
- return null;
- }
- TaskCompletionEvent[] tasks = job.getTaskCompletionEvents(0);
- // 0th even is setup, 1 event is launcher, 2 event is cleanup
- if (tasks != null && tasks.length == 3 && tasks[1] != null) {
- return tasks[1].getTaskTrackerHttp() + "/tasklog?attemptid="
- + tasks[1].getTaskAttemptId() + "&all=true";
- } else {
- LOG.warn("No running task for job: " + jobId);
- }
- return null;
- }
-
- private InputStream getURLinputStream(URL url) throws IOException {
- URLConnection connection = url.openConnection();
- connection.setDoOutput(true);
- connection.connect();
- return connection.getInputStream();
- }
+ private static final Logger LOG = Logger.getLogger(LogMover.class);
+
+ private static class ARGS {
+ String oozieUrl;
+ String subflowId;
+ String runId;
+ String logDir;
+ String status;
+ String entityType;
+ }
+
+ public static void main(String[] args) throws Exception {
+ ToolRunner.run(new LogMover(), args);
+ }
+
+ @Override
+ public int run(String[] arguments) throws Exception {
+ try {
+ ARGS args = new ARGS();
+ setupArgs(arguments, args);
+ OozieClient client = new OozieClient(args.oozieUrl);
+ WorkflowJob jobInfo = null;
+ try {
+ jobInfo = client.getJobInfo(args.subflowId);
+ } catch (OozieClientException e) {
+ LOG.error("Error getting jobinfo for: " + args.subflowId, e);
+ return 0;
+ }
+ Path path = new Path(args.logDir + "/"
+ + String.format("%03d", Integer.parseInt(args.runId)));
+
+ FileSystem fs = path.getFileSystem(getConf());
+
+ if (args.entityType.equalsIgnoreCase(EntityType.FEED.name())) {
+ // if replication wf
+ copyOozieLog(client, fs, path, jobInfo.getId());
+ copyTTlogs(args, fs, path, jobInfo.getActions().get(2));
+ } else {
+ // if process wf
+ String subflowId = jobInfo.getExternalId();
+ copyOozieLog(client, fs, path, subflowId);
+ WorkflowJob subflowInfo = client.getJobInfo(subflowId);
+ List<WorkflowAction> actions = subflowInfo.getActions();
+ for (WorkflowAction action : actions) {
+ if (action.getType().equals("pig")
+ || action.getType().equals("java")) {
+ copyTTlogs(args, fs, path, action);
+ } else {
+ LOG.info("Ignoring hadoop TT log for non-pig and non-java action:"
+ + action.getName());
+ }
+ }
+
+ }
+
+ } catch (Exception e) {
+ LOG.error("Exception in log mover:", e);
+ }
+ return 0;
+ }
+
+ private void copyOozieLog(OozieClient client, FileSystem fs, Path path,
+ String id) throws OozieClientException, IOException {
+ InputStream in = new ByteArrayInputStream(client.getJobLog(id).getBytes());
+ OutputStream out = fs.create(new Path(path, "oozie.log"));
+ IOUtils.copyBytes(in, out, 4096, true);
+ LOG.info("Copied oozie log to " + path);
+ }
+
+ private void copyTTlogs(ARGS args, FileSystem fs, Path path,
+ WorkflowAction action) throws Exception {
+ String ttLogURL = getTTlogURL(action.getExternalId());
+ if (ttLogURL != null) {
+ LOG.info("Fetching log for action: " + action.getExternalId()
+ + " from url: " + ttLogURL);
+ InputStream in = getURLinputStream(new URL(ttLogURL));
+ OutputStream out = fs.create(new Path(path, action.getName() + "_"
+ + getMappedStatus(action.getStatus()) + ".log"));
+ IOUtils.copyBytes(in, out, 4096, true);
+ LOG.info("Copied log to " + path);
+ }
+ }
+
+ private String getMappedStatus(WorkflowAction.Status status) {
+ if (status == WorkflowAction.Status.FAILED
+ || status == WorkflowAction.Status.KILLED
+ || status == WorkflowAction.Status.ERROR) {
+ return "FAILED";
+ } else {
+ return "SUCCEEDED";
+ }
+ }
+
+ private void setupArgs(String[] arguments, ARGS args) throws ParseException {
+ Options options = new Options();
+ Option opt;
+ opt = new Option("workflowEngineUrl", true,
+ "url of workflow engine, ex:oozie");
+ opt.setRequired(true);
+ options.addOption(opt);
+ opt = new Option("subflowId", true, "external id of userworkflow");
+ opt.setRequired(true);
+ options.addOption(opt);
+ opt = new Option("runId", true, "current workflow's runid");
+ opt.setRequired(true);
+ options.addOption(opt);
+ opt = new Option("logDir", true, "log dir where job logs are stored");
+ opt.setRequired(true);
+ options.addOption(opt);
+ opt = new Option("status", true, "user workflow status");
+ opt.setRequired(true);
+ options.addOption(opt);
+ opt = new Option("entityType", true, "entity type feed or process");
+ opt.setRequired(true);
+ options.addOption(opt);
+
+ CommandLine cmd = new GnuParser().parse(options, arguments);
+
+ args.oozieUrl = cmd.getOptionValue("workflowEngineUrl");
+ args.subflowId = cmd.getOptionValue("subflowId");
+ args.runId = cmd.getOptionValue("runId");
+ args.logDir = cmd.getOptionValue("logDir");
+ args.status = cmd.getOptionValue("status");
+ args.entityType = cmd.getOptionValue("entityType");
+
+ }
+
+ private String getTTlogURL(String jobId) throws Exception {
+ JobConf jobConf = new JobConf(getConf());
+ JobClient jobClient = new JobClient(jobConf);
+ RunningJob job = jobClient.getJob(JobID.forName(jobId));
+ if (job == null) {
+ LOG.warn("No running job for job id: " + jobId);
+ return null;
+ }
+ TaskCompletionEvent[] tasks = job.getTaskCompletionEvents(0);
+ // 0th even is setup, 1 event is launcher, 2 event is cleanup
+ if (tasks != null && tasks.length == 3 && tasks[1] != null) {
+ return tasks[1].getTaskTrackerHttp() + "/tasklog?attemptid="
+ + tasks[1].getTaskAttemptId() + "&all=true";
+ } else {
+ LOG.warn("No running task for job: " + jobId);
+ }
+ return null;
+ }
+
+ private InputStream getURLinputStream(URL url) throws IOException {
+ URLConnection connection = url.openConnection();
+ connection.setDoOutput(true);
+ connection.connect();
+ return connection.getInputStream();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/oozie/src/main/java/org/apache/falcon/logging/LogProvider.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/logging/LogProvider.java b/oozie/src/main/java/org/apache/falcon/logging/LogProvider.java
index 86b9b71..11eeadd 100644
--- a/oozie/src/main/java/org/apache/falcon/logging/LogProvider.java
+++ b/oozie/src/main/java/org/apache/falcon/logging/LogProvider.java
@@ -17,13 +17,7 @@
*/
package org.apache.falcon.logging;
-import java.io.IOException;
-
import org.apache.commons.lang.StringUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
import org.apache.falcon.FalconException;
import org.apache.falcon.entity.ClusterHelper;
import org.apache.falcon.entity.EntityUtil;
@@ -33,127 +27,133 @@ import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.resource.InstancesResult.Instance;
import org.apache.falcon.resource.InstancesResult.InstanceAction;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.apache.oozie.client.OozieClientException;
import org.mortbay.log.Log;
+import java.io.IOException;
+
public final class LogProvider {
- private static final Logger LOG = Logger.getLogger(LogProvider.class);
-
- public Instance populateLogUrls(Entity entity, Instance instance,
- String runId) throws FalconException {
-
- Cluster clusterObj = (Cluster) ConfigurationStore.get().get(
- EntityType.CLUSTER, instance.cluster);
- String resolvedRunId = "-";
- try {
- FileSystem fs = FileSystem.get(
- new Path(ClusterHelper.getStorageUrl(clusterObj)).toUri(),
- new Configuration());
- resolvedRunId = getResolvedRunId(fs, clusterObj, entity, instance,
- runId);
- // if runId param is not resolved, i.e job is killed or not started
- // or running
- if (resolvedRunId.equals("-")
- && StringUtils.isEmpty(instance.logFile)) {
- instance.logFile = "-";
- return instance;
- }
- return populateActionLogUrls(fs, clusterObj, entity, instance,
- resolvedRunId);
- } catch (IOException e) {
- LOG.warn("Exception while getting FS in LogProvider", e);
- } catch (Exception e) {
- LOG.warn("Exception in LogProvider while getting resolving run id",
- e);
- }
- return instance;
- }
-
- public String getResolvedRunId(FileSystem fs, Cluster cluster,
- Entity entity, Instance instance, String runId)
- throws FalconException, IOException {
- if (StringUtils.isEmpty(runId)) {
- Path jobPath = new Path(ClusterHelper.getStorageUrl(cluster),
- EntityUtil.getLogPath(cluster, entity) + "/job-"
- + EntityUtil.UTCtoURIDate(instance.instance) + "/*");
-
- FileStatus[] runs = fs.globStatus(jobPath);
- if (runs.length > 0) {
- // this is the latest run, dirs are sorted in increasing
- // order of runs
- return runs[runs.length - 1].getPath().getName();
- } else {
- LOG.warn("No run dirs are available in logs dir:" + jobPath);
- return "-";
- }
- } else {
- Path jobPath = new Path(ClusterHelper.getStorageUrl(cluster),
- EntityUtil.getLogPath(cluster, entity) + "/job-"
- + EntityUtil.UTCtoURIDate(instance.instance) + "/"
- + getFormatedRunId(runId));
- if (fs.exists(jobPath)) {
- return getFormatedRunId(runId);
- } else {
- Log.warn("No run dirs are available in logs dir:" + jobPath);
- return "-";
- }
- }
-
- }
-
- private Instance populateActionLogUrls(FileSystem fs, Cluster cluster,
- Entity entity, Instance instance, String formatedRunId)
- throws FalconException, OozieClientException, IOException {
-
- Path actionPaths = new Path(ClusterHelper.getStorageUrl(cluster),
- EntityUtil.getLogPath(cluster, entity) + "/job-"
- + EntityUtil.UTCtoURIDate(instance.instance) + "/"
- + formatedRunId + "/*");
- FileStatus[] actions = fs.globStatus(actionPaths);
- InstanceAction[] instanceActions = new InstanceAction[actions.length - 1];
- instance.actions = instanceActions;
- int i = 0;
- for (FileStatus file : actions) {
- Path filePath = file.getPath();
- String dfsBrowserUrl = getDFSbrowserUrl(
- ClusterHelper.getStorageUrl(cluster),
- EntityUtil.getLogPath(cluster, entity) + "/job-"
- + EntityUtil.UTCtoURIDate(instance.instance) + "/"
- + formatedRunId, file.getPath().getName());
- if (filePath.getName().equals("oozie.log")) {
- instance.logFile = dfsBrowserUrl;
- continue;
- }
-
- InstanceAction instanceAction = new InstanceAction(
- getActionName(filePath.getName()),
- getActionStatus(filePath.getName()), dfsBrowserUrl);
- instanceActions[i++] = instanceAction;
- }
-
- return instance;
-
- }
-
- private String getActionName(String fileName) {
- return fileName.replaceAll("_SUCCEEDED.log", "").replaceAll(
- "_FAILED.log", "");
- }
-
- private String getActionStatus(String fileName) {
- return fileName.contains("_SUCCEEDED.log") ? "SUCCEEDED" : "FAILED";
- }
-
- private String getDFSbrowserUrl(String hdfsPath, String logPath,
- String fileName) throws FalconException {
- String scheme = new Path(hdfsPath).toUri().getScheme();
- String httpUrl = hdfsPath.replaceAll(scheme+"://", "http://").replaceAll(
- ":[0-9]+", ":50070");
- return new Path(httpUrl, "/data/"+logPath + "/" + fileName).toString();
- }
-
- private String getFormatedRunId(String runId) {
- return String.format("%03d", Integer.parseInt(runId));
- }
+ private static final Logger LOG = Logger.getLogger(LogProvider.class);
+
+ public Instance populateLogUrls(Entity entity, Instance instance,
+ String runId) throws FalconException {
+
+ Cluster clusterObj = (Cluster) ConfigurationStore.get().get(
+ EntityType.CLUSTER, instance.cluster);
+ String resolvedRunId = "-";
+ try {
+ FileSystem fs = FileSystem.get(
+ new Path(ClusterHelper.getStorageUrl(clusterObj)).toUri(),
+ new Configuration());
+ resolvedRunId = getResolvedRunId(fs, clusterObj, entity, instance,
+ runId);
+ // if runId param is not resolved, i.e job is killed or not started
+ // or running
+ if (resolvedRunId.equals("-")
+ && StringUtils.isEmpty(instance.logFile)) {
+ instance.logFile = "-";
+ return instance;
+ }
+ return populateActionLogUrls(fs, clusterObj, entity, instance,
+ resolvedRunId);
+ } catch (IOException e) {
+ LOG.warn("Exception while getting FS in LogProvider", e);
+ } catch (Exception e) {
+ LOG.warn("Exception in LogProvider while getting resolving run id",
+ e);
+ }
+ return instance;
+ }
+
+ public String getResolvedRunId(FileSystem fs, Cluster cluster,
+ Entity entity, Instance instance, String runId)
+ throws FalconException, IOException {
+ if (StringUtils.isEmpty(runId)) {
+ Path jobPath = new Path(ClusterHelper.getStorageUrl(cluster),
+ EntityUtil.getLogPath(cluster, entity) + "/job-"
+ + EntityUtil.UTCtoURIDate(instance.instance) + "/*");
+
+ FileStatus[] runs = fs.globStatus(jobPath);
+ if (runs.length > 0) {
+ // this is the latest run, dirs are sorted in increasing
+ // order of runs
+ return runs[runs.length - 1].getPath().getName();
+ } else {
+ LOG.warn("No run dirs are available in logs dir:" + jobPath);
+ return "-";
+ }
+ } else {
+ Path jobPath = new Path(ClusterHelper.getStorageUrl(cluster),
+ EntityUtil.getLogPath(cluster, entity) + "/job-"
+ + EntityUtil.UTCtoURIDate(instance.instance) + "/"
+ + getFormatedRunId(runId));
+ if (fs.exists(jobPath)) {
+ return getFormatedRunId(runId);
+ } else {
+ Log.warn("No run dirs are available in logs dir:" + jobPath);
+ return "-";
+ }
+ }
+
+ }
+
+ private Instance populateActionLogUrls(FileSystem fs, Cluster cluster,
+ Entity entity, Instance instance, String formatedRunId)
+ throws FalconException, OozieClientException, IOException {
+
+ Path actionPaths = new Path(ClusterHelper.getStorageUrl(cluster),
+ EntityUtil.getLogPath(cluster, entity) + "/job-"
+ + EntityUtil.UTCtoURIDate(instance.instance) + "/"
+ + formatedRunId + "/*");
+ FileStatus[] actions = fs.globStatus(actionPaths);
+ InstanceAction[] instanceActions = new InstanceAction[actions.length - 1];
+ instance.actions = instanceActions;
+ int i = 0;
+ for (FileStatus file : actions) {
+ Path filePath = file.getPath();
+ String dfsBrowserUrl = getDFSbrowserUrl(
+ ClusterHelper.getStorageUrl(cluster),
+ EntityUtil.getLogPath(cluster, entity) + "/job-"
+ + EntityUtil.UTCtoURIDate(instance.instance) + "/"
+ + formatedRunId, file.getPath().getName());
+ if (filePath.getName().equals("oozie.log")) {
+ instance.logFile = dfsBrowserUrl;
+ continue;
+ }
+
+ InstanceAction instanceAction = new InstanceAction(
+ getActionName(filePath.getName()),
+ getActionStatus(filePath.getName()), dfsBrowserUrl);
+ instanceActions[i++] = instanceAction;
+ }
+
+ return instance;
+
+ }
+
+ private String getActionName(String fileName) {
+ return fileName.replaceAll("_SUCCEEDED.log", "").replaceAll(
+ "_FAILED.log", "");
+ }
+
+ private String getActionStatus(String fileName) {
+ return fileName.contains("_SUCCEEDED.log") ? "SUCCEEDED" : "FAILED";
+ }
+
+ private String getDFSbrowserUrl(String hdfsPath, String logPath,
+ String fileName) throws FalconException {
+ String scheme = new Path(hdfsPath).toUri().getScheme();
+ String httpUrl = hdfsPath.replaceAll(scheme + "://", "http://").replaceAll(
+ ":[0-9]+", ":50070");
+ return new Path(httpUrl, "/data/" + logPath + "/" + fileName).toString();
+ }
+
+ private String getFormatedRunId(String runId) {
+ return String.format("%03d", Integer.parseInt(runId));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/oozie/src/main/java/org/apache/falcon/service/FalconPathFilter.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/service/FalconPathFilter.java b/oozie/src/main/java/org/apache/falcon/service/FalconPathFilter.java
index 398369b..03bcea6 100644
--- a/oozie/src/main/java/org/apache/falcon/service/FalconPathFilter.java
+++ b/oozie/src/main/java/org/apache/falcon/service/FalconPathFilter.java
@@ -21,7 +21,7 @@ package org.apache.falcon.service;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
-public interface FalconPathFilter extends PathFilter{
-
+public interface FalconPathFilter extends PathFilter {
+
String getJarName(Path path);
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java b/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java
index de006f1..841177a 100644
--- a/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java
+++ b/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java
@@ -18,13 +18,6 @@
package org.apache.falcon.service;
-import java.io.File;
-import java.io.IOException;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
import org.apache.falcon.FalconException;
import org.apache.falcon.entity.ClusterHelper;
import org.apache.falcon.entity.v0.Entity;
@@ -33,26 +26,37 @@ import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.cluster.Interfacetype;
import org.apache.falcon.util.DeploymentUtil;
import org.apache.falcon.util.StartupProperties;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
+import java.io.File;
+import java.io.IOException;
+
public class SharedLibraryHostingService implements ConfigurationChangeListener {
private static Logger LOG = Logger.getLogger(SharedLibraryHostingService.class);
private static final String[] LIBS = StartupProperties.get().getProperty("shared.libs").split(",");
-
+
private static final FalconPathFilter nonFalconJarFilter = new FalconPathFilter() {
@Override
public boolean accept(Path path) {
- for(String jarName:LIBS) {
- if(path.getName().startsWith(jarName)) return true;
+ for (String jarName : LIBS) {
+ if (path.getName().startsWith(jarName)) {
+ return true;
+ }
}
return false;
}
@Override
public String getJarName(Path path) {
- for(String jarName:LIBS) {
- if(path.getName().startsWith(jarName)) return jarName;
+ for (String jarName : LIBS) {
+ if (path.getName().startsWith(jarName)) {
+ return jarName;
+ }
}
throw new IllegalArgumentException(path + " is not accepted!");
}
@@ -67,36 +71,41 @@ public class SharedLibraryHostingService implements ConfigurationChangeListener
}
}
- public static void pushLibsToHDFS(String path, Cluster cluster, FalconPathFilter pathFilter) throws IOException, FalconException {
+ public static void pushLibsToHDFS(String path, Cluster cluster, FalconPathFilter pathFilter)
+ throws IOException, FalconException {
String localPaths = StartupProperties.get().getProperty("system.lib.location");
assert localPaths != null && !localPaths.isEmpty() : "Invalid value for system.lib.location";
if (!new File(localPaths).isDirectory()) {
- throw new FalconException(localPaths + " configured for system.lib.location doesn't contain any valid libs");
+ throw new FalconException(
+ localPaths + " configured for system.lib.location doesn't contain any valid libs");
}
-
+
Configuration conf = ClusterHelper.getConfiguration(cluster);
conf.setInt("ipc.client.connect.max.retries", 10);
- FileSystem fs = null;
- try {
- fs = FileSystem.get(conf);
- } catch (Exception e) {
- throw new FalconException("Unable to connect to HDFS: "
- + ClusterHelper.getStorageUrl(cluster));
- }
+ FileSystem fs = null;
+ try {
+ fs = FileSystem.get(conf);
+ } catch (Exception e) {
+ throw new FalconException("Unable to connect to HDFS: "
+ + ClusterHelper.getStorageUrl(cluster));
+ }
Path clusterPath = new Path(path);
- if(!fs.exists(clusterPath))
+ if (!fs.exists(clusterPath)) {
fs.mkdirs(clusterPath);
-
+ }
+
for (File localFile : new File(localPaths).listFiles()) {
Path localPath = new Path(localFile.getAbsolutePath());
- if (!pathFilter.accept(localPath))
+ if (!pathFilter.accept(localPath)) {
continue;
+ }
Path clusterFile = new Path(path, pathFilter.getJarName(localPath) + ".jar");
if (fs.exists(clusterFile)) {
FileStatus fstat = fs.getFileStatus(clusterFile);
- if (fstat.getLen() == localFile.length())
+ if (fstat.getLen() == localFile.length()) {
continue;
+ }
}
fs.copyFromLocalFile(false, true, new Path(localFile.getAbsolutePath()), clusterFile);
LOG.info("Copied " + localFile.getAbsolutePath() + " to " + path + " in " + fs.getUri());
@@ -105,13 +114,15 @@ public class SharedLibraryHostingService implements ConfigurationChangeListener
@Override
public void onAdd(Entity entity) throws FalconException {
- if (entity.getEntityType() != EntityType.CLUSTER)
+ if (entity.getEntityType() != EntityType.CLUSTER) {
return;
+ }
Cluster cluster = (Cluster) entity;
String currentColo = DeploymentUtil.getCurrentColo();
- if (DeploymentUtil.isEmbeddedMode() || currentColo.equals(cluster.getColo()))
+ if (DeploymentUtil.isEmbeddedMode() || currentColo.equals(cluster.getColo())) {
addLibsTo(cluster);
+ }
}
@Override
@@ -121,14 +132,15 @@ public class SharedLibraryHostingService implements ConfigurationChangeListener
@Override
public void onChange(Entity oldEntity, Entity newEntity) throws FalconException {
- if (oldEntity.getEntityType() != EntityType.CLUSTER)
+ if (oldEntity.getEntityType() != EntityType.CLUSTER) {
return;
+ }
Cluster oldCluster = (Cluster) oldEntity;
Cluster newCluster = (Cluster) newEntity;
if (!ClusterHelper.getInterface(oldCluster, Interfacetype.WRITE).getEndpoint()
.equals(ClusterHelper.getInterface(newCluster, Interfacetype.WRITE).getEndpoint())
|| !ClusterHelper.getInterface(oldCluster, Interfacetype.WORKFLOW).getEndpoint()
- .equals(ClusterHelper.getInterface(newCluster, Interfacetype.WORKFLOW).getEndpoint())) {
+ .equals(ClusterHelper.getInterface(newCluster, Interfacetype.WORKFLOW).getEndpoint())) {
addLibsTo(newCluster);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/oozie/src/main/java/org/apache/falcon/util/OozieUtils.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/util/OozieUtils.java b/oozie/src/main/java/org/apache/falcon/util/OozieUtils.java
index 62afde1..20ef6bf 100644
--- a/oozie/src/main/java/org/apache/falcon/util/OozieUtils.java
+++ b/oozie/src/main/java/org/apache/falcon/util/OozieUtils.java
@@ -17,12 +17,12 @@
*/
package org.apache.falcon.util;
+import org.apache.hadoop.conf.Configuration;
+
import java.io.ByteArrayInputStream;
import java.util.Map;
import java.util.Properties;
-import org.apache.hadoop.conf.Configuration;
-
public final class OozieUtils {
public static Properties toProperties(String properties) {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java b/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
index fe41861..4c99bfc 100644
--- a/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
+++ b/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
@@ -17,192 +17,189 @@
*/
package org.apache.falcon.workflow;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.GnuParser;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.*;
+import org.apache.falcon.logging.LogMover;
+import org.apache.falcon.messaging.MessageProducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
-import org.apache.falcon.logging.LogMover;
-import org.apache.falcon.messaging.MessageProducer;
import org.apache.log4j.Logger;
-import org.mortbay.log.Log;
-
-public class FalconPostProcessing extends Configured implements Tool{
- private static final Logger LOG = Logger.getLogger(FalconPostProcessing.class);
-
- public enum Arg{
- CLUSTER("cluster","name of the current cluster"),
- ENTITY_TYPE("entityType","type of the entity"),
- ENTITY_NAME("entityName","name of the entity"),
- NOMINAL_TIME("nominalTime","instance time"),
- OPERATION("operation","operation like generate, delete, replicate"),
- WORKFLOW_ID("workflowId","current workflow-id of the instance"),
- RUN_ID("runId","current run-id of the instance"),
- STATUS("status","status of the user workflow isnstance"),
- TIMESTAMP("timeStamp","current timestamp"),
- TOPIC_NAME("topicName","name of the topic to be used to send JMS message"),
- BRKR_IMPL_CLASS("brokerImplClass","falcon message broker Implementation class"),
- BRKR_URL("brokerUrl","falcon message broker url"),
- USER_BRKR_IMPL_CLASS("userBrokerImplClass","user broker Impl class"),
- USER_BRKR_URL("userBrokerUrl","user broker url"),
- BRKR_TTL("brokerTTL", "time to live for broker message in sec"),
- FEED_NAMES("feedNames", "name of the feeds which are generated/replicated/deleted"),
- FEED_INSTANCE_PATHS("feedInstancePaths","comma seperated feed instance paths"),
- LOG_FILE("logFile","log file path where feeds to be deleted are recorded"),
- WF_ENGINE_URL("workflowEngineUrl","url of workflow engine server, ex:oozie"),
- USER_SUBFLOW_ID("subflowId","external id of user workflow"),
- LOG_DIR("logDir","log dir where job logs are copied");
-
- private String name;
- private String description;
- Arg(String name, String description){
- this.name=name;
- this.description=description;
- }
-
- public Option getOption() {
- return new Option(this.name, true, this.description);
- }
-
- public String getOptionName(){
- return this.name;
- }
-
- public String getOptionValue(CommandLine cmd){
- return cmd.getOptionValue(this.name);
- }
-
- }
-
- public static void main(String[] args) throws Exception {
- ToolRunner.run(new Configuration(), new FalconPostProcessing(), args);
- }
-
- @Override
- public int run(String[] args) throws Exception {
-
- CommandLine cmd = getCommand(args);
-
- LOG.info("Sending user message "+cmd);
- invokeUserMessageProducer(cmd);
- //LogMover doesnt throw exception, a failed logmover will not fail the user workflow
- LOG.info("Moving logs "+cmd);
- invokeLogProducer(cmd);
- LOG.info("Sending falcon message "+cmd);
- invokeFalconMessageProducer(cmd);
-
- return 0;
- }
-
- private void invokeUserMessageProducer(CommandLine cmd) throws Exception {
- List<String> args = new ArrayList<String>();
- addArg(args,cmd, Arg.CLUSTER);
- addArg(args,cmd, Arg.ENTITY_TYPE);
- addArg(args,cmd,Arg.ENTITY_NAME);
- addArg(args,cmd,Arg.NOMINAL_TIME);
- addArg(args,cmd,Arg.OPERATION);
- addArg(args,cmd,Arg.WORKFLOW_ID);
- addArg(args,cmd,Arg.RUN_ID);
- addArg(args,cmd,Arg.STATUS);
- addArg(args,cmd,Arg.TIMESTAMP);
- //special args for user JMS message producer
- args.add("-"+Arg.TOPIC_NAME.getOptionName()); //user topic
- args.add("FALCON."+Arg.ENTITY_NAME.getOptionValue(cmd));
- //note, the user broker impl class arg name to MessageProducer is brokerImplClass
- args.add("-"+Arg.BRKR_IMPL_CLASS.getOptionName());
- args.add(Arg.USER_BRKR_IMPL_CLASS.getOptionValue(cmd));
- args.add("-"+Arg.BRKR_URL.getOptionName());
- args.add(Arg.USER_BRKR_URL.getOptionValue(cmd));
- addArg(args,cmd,Arg.BRKR_TTL);
- addArg(args,cmd,Arg.FEED_NAMES);
- addArg(args,cmd,Arg.FEED_INSTANCE_PATHS);
- addArg(args,cmd,Arg.LOG_FILE);
-
- MessageProducer.main(args.toArray(new String[0]));
- }
-
- private void invokeFalconMessageProducer(CommandLine cmd) throws Exception {
- List<String> args = new ArrayList<String>();
- addArg(args,cmd, Arg.CLUSTER);
- addArg(args,cmd, Arg.ENTITY_TYPE);
- addArg(args,cmd,Arg.ENTITY_NAME);
- addArg(args,cmd,Arg.NOMINAL_TIME);
- addArg(args,cmd,Arg.OPERATION);
- addArg(args,cmd,Arg.WORKFLOW_ID);
- addArg(args,cmd,Arg.RUN_ID);
- addArg(args,cmd,Arg.STATUS);
- addArg(args,cmd,Arg.TIMESTAMP);
- //special args Falcon JMS message producer
- args.add("-"+Arg.TOPIC_NAME.getOptionName());
- args.add("FALCON.ENTITY.TOPIC");
- args.add("-"+Arg.BRKR_IMPL_CLASS.getOptionName());
- args.add(Arg.BRKR_IMPL_CLASS.getOptionValue(cmd));
- args.add("-"+Arg.BRKR_URL.getOptionName());
- args.add(Arg.BRKR_URL.getOptionValue(cmd));
- addArg(args,cmd,Arg.BRKR_TTL);
- addArg(args,cmd,Arg.FEED_NAMES);
- addArg(args,cmd,Arg.FEED_INSTANCE_PATHS);
- addArg(args,cmd,Arg.LOG_FILE);
-
- MessageProducer.main(args.toArray(new String[0]));
-
- }
- private void invokeLogProducer(CommandLine cmd) throws Exception{
- List<String> args = new ArrayList<String>();
- addArg(args,cmd,Arg.WF_ENGINE_URL);
- addArg(args,cmd,Arg.ENTITY_TYPE);
- addArg(args,cmd,Arg.USER_SUBFLOW_ID);
- addArg(args,cmd,Arg.RUN_ID);
- addArg(args,cmd,Arg.LOG_DIR);
- addArg(args,cmd,Arg.STATUS);
-
- LogMover.main(args.toArray(new String[0]));
-
- }
-
- private void addArg(List<String> args, CommandLine cmd, Arg arg) {
- args.add("-"+arg.getOptionName());
- args.add(arg.getOptionValue(cmd));
- }
-
- private static CommandLine getCommand(String[] arguments)
- throws ParseException {
- Options options = new Options();
- addOption(options, Arg.CLUSTER);
- addOption(options, Arg.ENTITY_TYPE);
- addOption(options, Arg.ENTITY_NAME);
- addOption(options, Arg.NOMINAL_TIME);
- addOption(options, Arg.OPERATION);
- addOption(options, Arg.WORKFLOW_ID);
- addOption(options, Arg.RUN_ID);
- addOption(options, Arg.STATUS);
- addOption(options, Arg.TIMESTAMP);
- addOption(options, Arg.BRKR_IMPL_CLASS);
- addOption(options, Arg.BRKR_URL);
- addOption(options, Arg.USER_BRKR_IMPL_CLASS);
- addOption(options, Arg.USER_BRKR_URL);
- addOption(options, Arg.BRKR_TTL);
- addOption(options, Arg.FEED_NAMES);
- addOption(options, Arg.FEED_INSTANCE_PATHS);
- addOption(options, Arg.LOG_FILE);
- addOption(options,Arg.WF_ENGINE_URL);
- addOption(options, Arg.USER_SUBFLOW_ID);
- addOption(options, Arg.LOG_DIR);
- return new GnuParser().parse(options, arguments);
- }
-
- private static void addOption(Options options, Arg arg) {
- Option option = arg.getOption();
- option.setRequired(true);
- options.addOption(option);
- }
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FalconPostProcessing extends Configured implements Tool {
+ private static final Logger LOG = Logger.getLogger(FalconPostProcessing.class);
+
+ public enum Arg {
+ CLUSTER("cluster", "name of the current cluster"),
+ ENTITY_TYPE("entityType", "type of the entity"),
+ ENTITY_NAME("entityName", "name of the entity"),
+ NOMINAL_TIME("nominalTime", "instance time"),
+ OPERATION("operation", "operation like generate, delete, replicate"),
+ WORKFLOW_ID("workflowId", "current workflow-id of the instance"),
+ RUN_ID("runId", "current run-id of the instance"),
+ STATUS("status", "status of the user workflow isnstance"),
+ TIMESTAMP("timeStamp", "current timestamp"),
+ TOPIC_NAME("topicName", "name of the topic to be used to send JMS message"),
+ BRKR_IMPL_CLASS("brokerImplClass", "falcon message broker Implementation class"),
+ BRKR_URL("brokerUrl", "falcon message broker url"),
+ USER_BRKR_IMPL_CLASS("userBrokerImplClass", "user broker Impl class"),
+ USER_BRKR_URL("userBrokerUrl", "user broker url"),
+ BRKR_TTL("brokerTTL", "time to live for broker message in sec"),
+ FEED_NAMES("feedNames", "name of the feeds which are generated/replicated/deleted"),
+ FEED_INSTANCE_PATHS("feedInstancePaths", "comma seperated feed instance paths"),
+ LOG_FILE("logFile", "log file path where feeds to be deleted are recorded"),
+ WF_ENGINE_URL("workflowEngineUrl", "url of workflow engine server, ex:oozie"),
+ USER_SUBFLOW_ID("subflowId", "external id of user workflow"),
+ LOG_DIR("logDir", "log dir where job logs are copied");
+
+ private String name;
+ private String description;
+
+ Arg(String name, String description) {
+ this.name = name;
+ this.description = description;
+ }
+
+ public Option getOption() {
+ return new Option(this.name, true, this.description);
+ }
+
+ public String getOptionName() {
+ return this.name;
+ }
+
+ public String getOptionValue(CommandLine cmd) {
+ return cmd.getOptionValue(this.name);
+ }
+
+ }
+
+ public static void main(String[] args) throws Exception {
+ ToolRunner.run(new Configuration(), new FalconPostProcessing(), args);
+ }
+
+ @Override
+ public int run(String[] args) throws Exception {
+
+ CommandLine cmd = getCommand(args);
+
+ LOG.info("Sending user message " + cmd);
+ invokeUserMessageProducer(cmd);
+ //LogMover doesnt throw exception, a failed logmover will not fail the user workflow
+ LOG.info("Moving logs " + cmd);
+ invokeLogProducer(cmd);
+ LOG.info("Sending falcon message " + cmd);
+ invokeFalconMessageProducer(cmd);
+
+ return 0;
+ }
+
+ private void invokeUserMessageProducer(CommandLine cmd) throws Exception {
+ List<String> args = new ArrayList<String>();
+ addArg(args, cmd, Arg.CLUSTER);
+ addArg(args, cmd, Arg.ENTITY_TYPE);
+ addArg(args, cmd, Arg.ENTITY_NAME);
+ addArg(args, cmd, Arg.NOMINAL_TIME);
+ addArg(args, cmd, Arg.OPERATION);
+ addArg(args, cmd, Arg.WORKFLOW_ID);
+ addArg(args, cmd, Arg.RUN_ID);
+ addArg(args, cmd, Arg.STATUS);
+ addArg(args, cmd, Arg.TIMESTAMP);
+ //special args for user JMS message producer
+ args.add("-" + Arg.TOPIC_NAME.getOptionName()); //user topic
+ args.add("FALCON." + Arg.ENTITY_NAME.getOptionValue(cmd));
+ //note, the user broker impl class arg name to MessageProducer is brokerImplClass
+ args.add("-" + Arg.BRKR_IMPL_CLASS.getOptionName());
+ args.add(Arg.USER_BRKR_IMPL_CLASS.getOptionValue(cmd));
+ args.add("-" + Arg.BRKR_URL.getOptionName());
+ args.add(Arg.USER_BRKR_URL.getOptionValue(cmd));
+ addArg(args, cmd, Arg.BRKR_TTL);
+ addArg(args, cmd, Arg.FEED_NAMES);
+ addArg(args, cmd, Arg.FEED_INSTANCE_PATHS);
+ addArg(args, cmd, Arg.LOG_FILE);
+
+ MessageProducer.main(args.toArray(new String[0]));
+ }
+
+ private void invokeFalconMessageProducer(CommandLine cmd) throws Exception {
+ List<String> args = new ArrayList<String>();
+ addArg(args, cmd, Arg.CLUSTER);
+ addArg(args, cmd, Arg.ENTITY_TYPE);
+ addArg(args, cmd, Arg.ENTITY_NAME);
+ addArg(args, cmd, Arg.NOMINAL_TIME);
+ addArg(args, cmd, Arg.OPERATION);
+ addArg(args, cmd, Arg.WORKFLOW_ID);
+ addArg(args, cmd, Arg.RUN_ID);
+ addArg(args, cmd, Arg.STATUS);
+ addArg(args, cmd, Arg.TIMESTAMP);
+ //special args Falcon JMS message producer
+ args.add("-" + Arg.TOPIC_NAME.getOptionName());
+ args.add("FALCON.ENTITY.TOPIC");
+ args.add("-" + Arg.BRKR_IMPL_CLASS.getOptionName());
+ args.add(Arg.BRKR_IMPL_CLASS.getOptionValue(cmd));
+ args.add("-" + Arg.BRKR_URL.getOptionName());
+ args.add(Arg.BRKR_URL.getOptionValue(cmd));
+ addArg(args, cmd, Arg.BRKR_TTL);
+ addArg(args, cmd, Arg.FEED_NAMES);
+ addArg(args, cmd, Arg.FEED_INSTANCE_PATHS);
+ addArg(args, cmd, Arg.LOG_FILE);
+
+ MessageProducer.main(args.toArray(new String[0]));
+
+ }
+
+ private void invokeLogProducer(CommandLine cmd) throws Exception {
+ List<String> args = new ArrayList<String>();
+ addArg(args, cmd, Arg.WF_ENGINE_URL);
+ addArg(args, cmd, Arg.ENTITY_TYPE);
+ addArg(args, cmd, Arg.USER_SUBFLOW_ID);
+ addArg(args, cmd, Arg.RUN_ID);
+ addArg(args, cmd, Arg.LOG_DIR);
+ addArg(args, cmd, Arg.STATUS);
+
+ LogMover.main(args.toArray(new String[0]));
+
+ }
+
+ private void addArg(List<String> args, CommandLine cmd, Arg arg) {
+ args.add("-" + arg.getOptionName());
+ args.add(arg.getOptionValue(cmd));
+ }
+
+ private static CommandLine getCommand(String[] arguments)
+ throws ParseException {
+ Options options = new Options();
+ addOption(options, Arg.CLUSTER);
+ addOption(options, Arg.ENTITY_TYPE);
+ addOption(options, Arg.ENTITY_NAME);
+ addOption(options, Arg.NOMINAL_TIME);
+ addOption(options, Arg.OPERATION);
+ addOption(options, Arg.WORKFLOW_ID);
+ addOption(options, Arg.RUN_ID);
+ addOption(options, Arg.STATUS);
+ addOption(options, Arg.TIMESTAMP);
+ addOption(options, Arg.BRKR_IMPL_CLASS);
+ addOption(options, Arg.BRKR_URL);
+ addOption(options, Arg.USER_BRKR_IMPL_CLASS);
+ addOption(options, Arg.USER_BRKR_URL);
+ addOption(options, Arg.BRKR_TTL);
+ addOption(options, Arg.FEED_NAMES);
+ addOption(options, Arg.FEED_INSTANCE_PATHS);
+ addOption(options, Arg.LOG_FILE);
+ addOption(options, Arg.WF_ENGINE_URL);
+ addOption(options, Arg.USER_SUBFLOW_ID);
+ addOption(options, Arg.LOG_DIR);
+ return new GnuParser().parse(options, arguments);
+ }
+
+ private static void addOption(Options options, Arg arg) {
+ Option option = arg.getOption();
+ option.setRequired(true);
+ options.addOption(option);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/oozie/src/main/java/org/apache/falcon/workflow/OozieWorkflowBuilder.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/workflow/OozieWorkflowBuilder.java b/oozie/src/main/java/org/apache/falcon/workflow/OozieWorkflowBuilder.java
index 85c8697..8d67a9b 100644
--- a/oozie/src/main/java/org/apache/falcon/workflow/OozieWorkflowBuilder.java
+++ b/oozie/src/main/java/org/apache/falcon/workflow/OozieWorkflowBuilder.java
@@ -18,11 +18,6 @@
package org.apache.falcon.workflow;
-import java.util.Date;
-import java.util.List;
-import java.util.Properties;
-
-import org.apache.hadoop.fs.Path;
import org.apache.falcon.FalconException;
import org.apache.falcon.entity.ClusterHelper;
import org.apache.falcon.entity.EntityUtil;
@@ -32,22 +27,27 @@ import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.cluster.Property;
import org.apache.falcon.workflow.engine.OozieWorkflowEngine;
+import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.apache.oozie.client.OozieClient;
+import java.util.Date;
+import java.util.List;
+import java.util.Properties;
+
public abstract class OozieWorkflowBuilder<T extends Entity> extends WorkflowBuilder<T> {
-
+
private static Logger LOG = Logger.getLogger(OozieWorkflowBuilder.class);
protected static final ConfigurationStore configStore = ConfigurationStore.get();
-
+
protected Properties createAppProperties(String clusterName, Path bundlePath, String user) throws FalconException {
Cluster cluster = EntityUtil.getEntity(EntityType.CLUSTER, clusterName);
Properties properties = new Properties();
- if (cluster.getProperties() != null) {
- addClusterProperties(properties, cluster.getProperties()
- .getProperties());
- }
+ if (cluster.getProperties() != null) {
+ addClusterProperties(properties, cluster.getProperties()
+ .getProperties());
+ }
properties.setProperty(OozieWorkflowEngine.NAME_NODE,
ClusterHelper.getStorageUrl(cluster));
properties.setProperty(OozieWorkflowEngine.JOB_TRACKER,
@@ -55,20 +55,20 @@ public abstract class OozieWorkflowBuilder<T extends Entity> extends WorkflowBui
properties.setProperty(OozieClient.BUNDLE_APP_PATH,
"${" + OozieWorkflowEngine.NAME_NODE + "}" + bundlePath.toString());
properties.setProperty("colo.name", cluster.getColo());
-
+
properties.setProperty(OozieClient.USER_NAME, user);
properties.setProperty(OozieClient.USE_SYSTEM_LIBPATH, "true");
properties.setProperty("falcon.libpath", ClusterHelper.getLocation(cluster, "working") + "/lib");
LOG.info("Cluster: " + cluster.getName() + ", PROPS: " + properties);
return properties;
}
-
- private void addClusterProperties(Properties properties,
- List<Property> clusterProperties) {
- for (Property prop : clusterProperties) {
- properties.setProperty(prop.getName(), prop.getValue());
- }
- }
- public abstract Date getNextStartTime(T entity, String cluster, Date now) throws FalconException;
+ private void addClusterProperties(Properties properties,
+ List<Property> clusterProperties) {
+ for (Property prop : clusterProperties) {
+ properties.setProperty(prop.getName(), prop.getValue());
+ }
+ }
+
+ public abstract Date getNextStartTime(T entity, String cluster, Date now) throws FalconException;
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a4d79f0c/oozie/src/main/java/org/apache/falcon/workflow/engine/NullBundleJob.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/workflow/engine/NullBundleJob.java b/oozie/src/main/java/org/apache/falcon/workflow/engine/NullBundleJob.java
index 985b297..f2c9ac3 100644
--- a/oozie/src/main/java/org/apache/falcon/workflow/engine/NullBundleJob.java
+++ b/oozie/src/main/java/org/apache/falcon/workflow/engine/NullBundleJob.java
@@ -18,13 +18,13 @@
package org.apache.falcon.workflow.engine;
-import java.util.Date;
-import java.util.List;
-
import org.apache.oozie.client.BundleJob;
import org.apache.oozie.client.CoordinatorJob;
-public class NullBundleJob implements BundleJob{
+import java.util.Date;
+import java.util.List;
+
+public class NullBundleJob implements BundleJob {
@Override
public String getAppPath() {