rebased and resolved the conflicts from master

apache · Aug 5, 2016 · 1a4dcd2 · 1a4dcd2
2 parents 271318b + 7d9687b
commit 1a4dcd2
Show file tree

Hide file tree

Showing 32 changed files with 336 additions and 128 deletions.
diff --git a/addons/hivedr/src/main/java/org/apache/falcon/hive/HiveDRTool.java b/addons/hivedr/src/main/java/org/apache/falcon/hive/HiveDRTool.java
@@ -136,13 +136,6 @@ private void init(String[] args) throws Exception {
         inputOptions = parseOptions(args);
         LOG.info("Input Options: {}", inputOptions);
 
-        // Update the source staging path
-        inputOptions.setSourceStagingPath();
-        inputOptions.setTargetStagingPath();
-
-        LOG.info("srcStaginPath: {}", inputOptions.getSourceStagingPath());
-        LOG.info("tgtStaginPath: {}", inputOptions.getTargetStagingPath());
-
         Configuration sourceConf = FileUtils.getConfiguration(getConf(), inputOptions.getSourceWriteEP(),
                 inputOptions.getSourceNNKerberosPrincipal());
         sourceClusterFS = FileSystem.get(sourceConf);
@@ -155,6 +148,14 @@ private void init(String[] args) throws Exception {
 
         // init DR status store
         drStore = new HiveDRStatusStore(targetClusterFS);
+
+        // Update the source staging path after initing DR status store
+        inputOptions.setSourceStagingPath();
+        inputOptions.setTargetStagingPath();
+
+        LOG.info("srcStaginPath: {}", inputOptions.getSourceStagingPath());
+        LOG.info("tgtStaginPath: {}", inputOptions.getTargetStagingPath());
+
         eventSoucerUtil = new EventSourcerUtils(jobConf, inputOptions.shouldKeepHistory(), inputOptions.getJobName());
     }
 
@@ -310,7 +311,7 @@ private String getLastEvents(Configuration conf) throws Exception {
     }
 
     private Map<String, Long> getLastDBTableEvents(Path lastEventIdFile) throws Exception {
-        Map<String, Long> lastEventsIdMap = new HashMap<String, Long>();
+        Map<String, Long> lastEventsIdMap = new HashMap<>();
         BufferedReader in = new BufferedReader(new InputStreamReader(jobFS.open(lastEventIdFile)));
         try {
             String line;

diff --git a/addons/hivedr/src/main/java/org/apache/falcon/hive/util/DRStatusStore.java b/addons/hivedr/src/main/java/org/apache/falcon/hive/util/DRStatusStore.java
@@ -30,7 +30,7 @@
  */
 public abstract class DRStatusStore {
 
-    public static final String BASE_DEFAULT_STORE_PATH = "/apps/data-mirroring/";
+    public static final String BASE_DEFAULT_STORE_PATH = "/apps/falcon/extensions/mirroring";
     public static final FsPermission DEFAULT_STORE_PERMISSION =
             new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.NONE);
 

diff --git a/addons/hivedr/src/main/java/org/apache/falcon/hive/util/EventUtils.java b/addons/hivedr/src/main/java/org/apache/falcon/hive/util/EventUtils.java
@@ -50,7 +50,7 @@
 import java.util.Properties;
 
 /**
- * Utility class to handle Hive events for data-mirroring.
+ * Utility class to handle Hive events for hive-mirroring.
  */
 public class EventUtils {
     private static final String DRIVER_NAME = "org.apache.hive.jdbc.HiveDriver";

diff --git a/common/src/main/resources/runtime.properties b/common/src/main/resources/runtime.properties
@@ -51,4 +51,4 @@
 
 *.falcon.service.ProxyUserService.proxyuser.#USER#.groups=*
 
-######### Proxyuser Configuration End #########
+######### Proxyuser Configuration End #########
diff --git a/common/src/main/resources/startup.properties b/common/src/main/resources/startup.properties
@@ -83,6 +83,7 @@
 
 ##### Workflow Job Execution Completion listeners #####
 *.workflow.execution.listeners=
+#org.apache.falcon.service.LogMoverService
 
 ######### Implementation classes #########
 
@@ -336,3 +337,9 @@ it.workflow.execution.listeners=org.apache.falcon.catalog.CatalogPartitionHandle
 # Backlog Metric Properties
 #*.falcon.backlog.metricservice.emit.interval.millisecs=60000
 #*.falcon.backlog.metricservice.recheck.interval.millisecs=600000
+
+# Property to remove postProcessing
+*.falcon.postprocessing.enable=true
+
+### LogMoveService Thread count
+*.falcon.logMoveService.threadCount=50
diff --git a/docs/src/site/twiki/Configuration.twiki b/docs/src/site/twiki/Configuration.twiki
@@ -317,6 +317,20 @@ su - $OOZIE_USER
 Where $OOZIE_USER is the Oozie user. For example, oozie.
 </verbatim>
 
+---+++Disabling Falcon Post Processing
+Falcon post processing performs two tasks:
+They send user notifications to Active mq.
+It moves oozie executor logs once the workflow finishes.
+
+If post processing is failing because of any reason user mind end up having a backlog in the pipeline thats why it has been made optional.
+
+To disable post processing set the following property to false in startup.properties :
+<verbatim>
+*.falcon.postprocessing.enable=false
+*.workflow.execution.listeners=org.apache.falcon.service.LogMoverService
+</verbatim>
+*NOTE : Please make sure Oozie JMS Notifications are enabled as logMoverService depends on the Oozie JMS Notification.*
+
 
 ---+++Enabling Falcon Native Scheudler
 You can either choose to schedule entities using Oozie's coordinator or using Falcon's native scheduler. To be able to

diff --git a/extensions/src/main/java/org/apache/falcon/extensions/store/ExtensionStore.java b/extensions/src/main/java/org/apache/falcon/extensions/store/ExtensionStore.java
@@ -151,16 +151,14 @@ public String getExtensionLibPath(final String extensionName) throws StoreAccess
 
             for (FileStatus fileStatus : files) {
                 if (fileStatus.getPath().getName().equalsIgnoreCase(LIBS_DIR)) {
-                    if (fileStatus.getLen() != 0) {
                         libsPath = Path.getPathWithoutSchemeAndAuthority(fileStatus.getPath());
-                    }
                     break;
                 }
             }
 
             if (libsPath == null) {
                 LOG.info("For extension " + extensionName + " there is no "
-                        + LIBS_DIR + "at the extension store path " + storePath);
+                        + LIBS_DIR + "at the extension store path " + extensionPath);
                 return null;
             } else {
                 return libsPath.toString();

diff --git a/extensions/src/test/resources/hive-mirroring-template.xml b/extensions/src/test/resources/hive-mirroring-template.xml
@@ -37,8 +37,8 @@
         <property name="oozie.wf.subworkflow.classpath.inheritance" value="true"/>
     </properties>
 
-    <workflow name="##workflow.name##" engine="oozie"
-              path="/apps/data-mirroring/workflows/hive-disaster-recovery-workflow.xml" lib="##workflow.lib.path##"/>
+    <workflow name="##workflow.name##" engine="oozie" path="/apps/falcon/extensions/hive-mirroring/resources/runtime/hive-mirroring-workflow.xml"
+    lib="##workflow.lib.path##"/>
     <retry policy="##retry.policy##" delay="##retry.delay##" attempts="3"/>
     <notification type="##notification.type##" to="##notification.receivers##"/>
     <ACL/>

diff --git a/falcon-regression/merlin/src/test/resources/HdfsRecipe/hdfs-replication-template.xml b/falcon-regression/merlin/src/test/resources/HdfsRecipe/hdfs-replication-template.xml
@@ -37,7 +37,7 @@
         <property name="oozie.wf.subworkflow.classpath.inheritance" value="true"/>
     </properties>
 
-    <workflow name="##falcon.recipe.workflow.name##" engine="oozie" path="/apps/data-mirroring/workflows/hdfs-replication-workflow.xml" lib="##workflow.lib.path##"/>
+    <workflow name="##falcon.recipe.workflow.name##" engine="oozie" path="/apps/falcon/extensions/hdfs-mirroring/resources/runtime/hdfs-mirroring-workflow.xml" lib="##workflow.lib.path##"/>
     <retry policy="##falcon.recipe.retry.policy##" delay="##falcon.recipe.retry.delay##" attempts="3"/>
     <notification type="##falcon.recipe.notification.type##" to="##falcon.recipe.notification.receivers##"/>
     <ACL/>

diff --git a/falcon-regression/merlin/src/test/resources/HdfsRecipe/hdfs-replication.properties b/falcon-regression/merlin/src/test/resources/HdfsRecipe/hdfs-replication.properties
@@ -25,7 +25,7 @@ falcon.recipe.name=hdfs-replication-monthly
 ##### Workflow properties
 falcon.recipe.workflow.name=hdfs-dr-workflow
 # Provide Wf absolute path. This can be HDFS or local FS path. If WF is on local FS it will be copied to HDFS
-falcon.recipe.workflow.path=/apps/data-mirroring/workflows/hdfs-replication-workflow.xml
+falcon.recipe.workflow.path=/apps/falcon/extensions/hdfs-mirroring/resources/runtime/hdfs-mirroring-workflow.xml
 
 ##### Cluster properties
 

diff --git a/falcon-regression/merlin/src/test/resources/HiveDrRecipe/hive-disaster-recovery-template.xml b/falcon-regression/merlin/src/test/resources/HiveDrRecipe/hive-disaster-recovery-template.xml
@@ -38,7 +38,7 @@
     </properties>
 
     <workflow name="##workflow.name##" engine="oozie"
-              path="/apps/data-mirroring/workflows/hive-disaster-recovery-workflow.xml" lib="##workflow.lib.path##"/>
+              path="/apps/falcon/extensions/hive-mirroring/resources/runtime/hive-mirroring-workflow.xml" lib="##workflow.lib.path##"/>
     <retry policy="##retry.policy##" delay="##retry.delay##" attempts="3"/>
     <notification type="##falcon.recipe.notification.type##" to="##falcon.recipe.notification.receivers##"/>
     <ACL/>

diff --git a/...n/merlin/src/test/resources/HiveDrSecureRecipe/hive-disaster-recovery-secure-template.xml b/...n/merlin/src/test/resources/HiveDrSecureRecipe/hive-disaster-recovery-secure-template.xml
@@ -38,7 +38,7 @@
     </properties>
 
     <workflow name="##workflow.name##" engine="oozie"
-              path="/apps/data-mirroring/workflows/hive-disaster-recovery-workflow.xml" lib="##workflow.lib.path##"/>
+              path="/apps/falcon/extensions/hive-mirroring/resources/runtime/hive-mirroring-workflow.xml" lib="##workflow.lib.path##"/>
     <retry policy="##retry.policy##" delay="##retry.delay##" attempts="3"/>
     <notification type="##falcon.recipe.notification.type##" to="##falcon.recipe.notification.receivers##"/>
     <ACL/>

diff --git a/falcon-ui/app/js/services/entity/entity-model.js b/falcon-ui/app/js/services/entity/entity-model.js
@@ -302,7 +302,7 @@
           workflow: {
             _name: "hdfs-dr-workflow",
             _engine: "oozie",
-            _path: "/apps/data-mirroring/workflows/hdfs-replication-workflow.xml",
+            _path: "/apps/falcon/extensions/hdfs-mirroring/resources/runtime/hdfs-mirroring-workflow.xml",
             _lib: ""
           },
           retry: {
@@ -434,7 +434,7 @@
           workflow: {
             _name: "falcon-dr-hive-workflow",
             _engine: "oozie",
-            _path: "/apps/data-mirroring/workflows/hive-disaster-recovery-workflow.xml",
+            _path: "/apps/falcon/extensions/hive-mirroring/resources/runtime/hive-mirroring-workflow.xml",
             _lib: ""
           },
           retry: {

diff --git a/...main/java/org/apache/falcon/lifecycle/engine/oozie/retention/AgeBasedWorkflowBuilder.java b/...main/java/org/apache/falcon/lifecycle/engine/oozie/retention/AgeBasedWorkflowBuilder.java
@@ -59,19 +59,27 @@ public static Properties build(Cluster cluster, Path basePath, Feed feed) throws
 
         //Add eviction action
         ACTION eviction = OozieBuilderUtils.unmarshalAction(EVICTION_ACTION_TEMPLATE);
-        OozieBuilderUtils.addTransition(eviction, OozieBuilderUtils.SUCCESS_POSTPROCESS_ACTION_NAME,
-                OozieBuilderUtils.FAIL_POSTPROCESS_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(eviction);
-
-        //Add post-processing actions
-        ACTION success = OozieBuilderUtils.getSuccessPostProcessAction();
-        OozieBuilderUtils.addTransition(success, OozieBuilderUtils.OK_ACTION_NAME, OozieBuilderUtils.FAIL_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(success);
-
-        ACTION fail = OozieBuilderUtils.getFailPostProcessAction();
-        OozieBuilderUtils.addTransition(fail, OozieBuilderUtils.FAIL_ACTION_NAME, OozieBuilderUtils.FAIL_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(fail);
 
+        if (!Boolean.parseBoolean(OozieBuilderUtils.ENABLE_POSTPROCESSING)){
+            OozieBuilderUtils.addTransition(eviction, OozieBuilderUtils.OK_ACTION_NAME,
+                    OozieBuilderUtils.FAIL_ACTION_NAME);
+            workflow.getDecisionOrForkOrJoin().add(eviction);
+        } else {
+            OozieBuilderUtils.addTransition(eviction, OozieBuilderUtils.SUCCESS_POSTPROCESS_ACTION_NAME,
+                    OozieBuilderUtils.FAIL_POSTPROCESS_ACTION_NAME);
+            workflow.getDecisionOrForkOrJoin().add(eviction);
+
+            //Add post-processing actions
+            ACTION success = OozieBuilderUtils.getSuccessPostProcessAction();
+            OozieBuilderUtils.addTransition(success, OozieBuilderUtils.OK_ACTION_NAME,
+                    OozieBuilderUtils.FAIL_ACTION_NAME);
+            workflow.getDecisionOrForkOrJoin().add(success);
+
+            ACTION fail = OozieBuilderUtils.getFailPostProcessAction();
+            OozieBuilderUtils.addTransition(fail, OozieBuilderUtils.FAIL_ACTION_NAME,
+                    OozieBuilderUtils.FAIL_ACTION_NAME);
+            workflow.getDecisionOrForkOrJoin().add(fail);
+        }
         OozieBuilderUtils.decorateWorkflow(workflow, wfName, EVICTION_ACTION_NAME);
         OozieBuilderUtils.addLibExtensionsToWorkflow(cluster, workflow, Tag.RETENTION, EntityType.FEED);
 

diff --git a/...cycle/src/main/java/org/apache/falcon/lifecycle/engine/oozie/utils/OozieBuilderUtils.java b/...cycle/src/main/java/org/apache/falcon/lifecycle/engine/oozie/utils/OozieBuilderUtils.java
@@ -102,6 +102,8 @@ public final class OozieBuilderUtils {
     public static final String ENTITY_PATH = "ENTITY_PATH";
     public static final String ENTITY_NAME = "ENTITY_NAME";
     public static final String IGNORE = "IGNORE";
+    public static final String ENABLE_POSTPROCESSING = StartupProperties.get().
+            getProperty("falcon.postprocessing.enable");
 
 
     static {

diff --git a/oozie/pom.xml b/oozie/pom.xml
@@ -105,7 +105,6 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
         </dependency>
-
     </dependencies>
 
     <build>

diff --git a/oozie/src/main/java/org/apache/falcon/logging/JobLogMover.java b/oozie/src/main/java/org/apache/falcon/logging/JobLogMover.java
@@ -27,6 +27,7 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.oozie.client.OozieClient;
 import org.apache.oozie.client.OozieClientException;
@@ -68,6 +69,19 @@ private Configuration getConf() {
         return conf == null ? new Configuration(): conf;
     }
 
+    public void moveLog(WorkflowExecutionContext context){
+        if (UserGroupInformation.isSecurityEnabled()) {
+            LOG.info("Unable to move logs as security is enabled.");
+            return;
+        }
+        try {
+            run(context);
+        } catch (Exception ignored) {
+            // Mask exception, a failed log mover will not fail the user workflow
+            LOG.error("Exception in job log mover:", ignored);
+        }
+    }
+
     public int run(WorkflowExecutionContext context) {
         try {
             OozieClient client = new OozieClient(context.getWorkflowEngineUrl());

diff --git a/oozie/src/main/java/org/apache/falcon/oozie/DatabaseExportWorkflowBuilder.java b/oozie/src/main/java/org/apache/falcon/oozie/DatabaseExportWorkflowBuilder.java
@@ -67,18 +67,7 @@ protected Properties getWorkflow(Cluster cluster, WORKFLOWAPP workflow, Path bui
         ImportExportCommon.addHCatalogProperties(props, entity, cluster, workflow, this, buildPath, sqoopExport);
         OozieUtils.marshalSqoopAction(action, actionJaxbElement);
 
-        addTransition(action, SUCCESS_POSTPROCESS_ACTION_NAME, FAIL_POSTPROCESS_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(action);
-
-        //Add post-processing actions
-        ACTION success = getSuccessPostProcessAction();
-        addTransition(success, OK_ACTION_NAME, FAIL_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(success);
-
-        ACTION fail = getFailPostProcessAction();
-        addTransition(fail, FAIL_ACTION_NAME, FAIL_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(fail);
-
+        addPostProcessing(workflow, action);
         decorateWorkflow(workflow, workflow.getName(), EXPORT_ACTION_NAME);
         addLibExtensionsToWorkflow(cluster, workflow, Tag.EXPORT);
 

diff --git a/oozie/src/main/java/org/apache/falcon/oozie/DatabaseImportWorkflowBuilder.java b/oozie/src/main/java/org/apache/falcon/oozie/DatabaseImportWorkflowBuilder.java
@@ -69,18 +69,7 @@ protected Properties getWorkflow(Cluster cluster, WORKFLOWAPP workflow, Path bui
         ImportExportCommon.addHCatalogProperties(props, entity, cluster, workflow, this, buildPath, sqoopImport);
         OozieUtils.marshalSqoopAction(action, actionJaxbElement);
 
-        addTransition(action, SUCCESS_POSTPROCESS_ACTION_NAME, FAIL_POSTPROCESS_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(action);
-
-        //Add post-processing actions
-        ACTION success = getSuccessPostProcessAction();
-        addTransition(success, OK_ACTION_NAME, FAIL_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(success);
-
-        ACTION fail = getFailPostProcessAction();
-        addTransition(fail, FAIL_ACTION_NAME, FAIL_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(fail);
-
+        addPostProcessing(workflow, action);
         decorateWorkflow(workflow, workflow.getName(), IMPORT_ACTION_NAME);
         addLibExtensionsToWorkflow(cluster, workflow, Tag.IMPORT);
 

diff --git a/oozie/src/main/java/org/apache/falcon/oozie/OozieOrchestrationWorkflowBuilder.java b/oozie/src/main/java/org/apache/falcon/oozie/OozieOrchestrationWorkflowBuilder.java
@@ -80,6 +80,13 @@
 public abstract class OozieOrchestrationWorkflowBuilder<T extends Entity> extends OozieEntityBuilder<T> {
     public static final String HIVE_CREDENTIAL_NAME = "falconHiveAuth";
 
+    public String getEnablePostProcessing() {
+        return enablePostprocessing;
+    }
+
+    private String enablePostprocessing = StartupProperties.get().
+            getProperty("falcon.postprocessing.enable");
+
     protected static final String USER_ACTION_NAME = "user-action";
     protected static final String PREPROCESS_ACTION_NAME = "pre-processing";
     protected static final String SUCCESS_POSTPROCESS_ACTION_NAME = "succeeded-post-processing";
@@ -130,6 +137,10 @@ public static OozieOrchestrationWorkflowBuilder get(Entity entity, Cluster clust
         return get(entity, cluster, lifecycle, Scheduler.OOZIE);
     }
 
+    public Boolean isPostProcessingEnabled(){
+        return Boolean.parseBoolean(getEnablePostProcessing());
+    }
+
     public static OozieOrchestrationWorkflowBuilder get(Entity entity, Cluster cluster, Tag lifecycle,
                                                         Scheduler scheduler)
         throws FalconException {
@@ -219,6 +230,25 @@ protected void decorateWorkflow(WORKFLOWAPP wf, String name, String startAction)
         wf.getDecisionOrForkOrJoin().add(kill);
     }
 
+    protected void addPostProcessing(WORKFLOWAPP workflow, ACTION action) throws FalconException{
+        if (!isPostProcessingEnabled()){
+            addTransition(action, OK_ACTION_NAME, FAIL_ACTION_NAME);
+            workflow.getDecisionOrForkOrJoin().add(action);
+        }else{
+            addTransition(action, SUCCESS_POSTPROCESS_ACTION_NAME, FAIL_POSTPROCESS_ACTION_NAME);
+            workflow.getDecisionOrForkOrJoin().add(action);
+
+            //Add post-processing actions
+            ACTION success = getSuccessPostProcessAction();
+            addTransition(success, OK_ACTION_NAME, FAIL_ACTION_NAME);
+            workflow.getDecisionOrForkOrJoin().add(success);
+
+            ACTION fail = getFailPostProcessAction();
+            addTransition(fail, FAIL_ACTION_NAME, FAIL_ACTION_NAME);
+            workflow.getDecisionOrForkOrJoin().add(fail);
+        }
+    }
+
     protected ACTION getSuccessPostProcessAction() throws FalconException {
         ACTION action = unmarshalAction(POSTPROCESS_TEMPLATE);
         decorateWithOozieRetries(action);

diff --git a/oozie/src/main/java/org/apache/falcon/oozie/feed/FSReplicationWorkflowBuilder.java b/oozie/src/main/java/org/apache/falcon/oozie/feed/FSReplicationWorkflowBuilder.java
@@ -59,20 +59,7 @@ public FSReplicationWorkflowBuilder(Feed entity) {
         addAdditionalReplicationProperties(replication);
         enableCounters(replication);
         enableTDE(replication);
-        addTransition(replication, SUCCESS_POSTPROCESS_ACTION_NAME, FAIL_POSTPROCESS_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(replication);
-
-        //Add post-processing actions
-        ACTION success = getSuccessPostProcessAction();
-        addHDFSServersConfig(success, src, target);
-        addTransition(success, OK_ACTION_NAME, FAIL_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(success);
-
-        ACTION fail = getFailPostProcessAction();
-        addHDFSServersConfig(fail, src, target);
-        addTransition(fail, FAIL_ACTION_NAME, FAIL_ACTION_NAME);
-        workflow.getDecisionOrForkOrJoin().add(fail);
-
+        addPostProcessing(workflow, replication);
         decorateWorkflow(workflow, wfName, start);
         return workflow;
     }