falcon-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From shaikid...@apache.org
Subject git commit: FALCON-38 Falcon's parent workflow actions (pre-processing & post-processing) should have multiple retries. Contributed by shaik.idris
Date Tue, 14 Jan 2014 16:46:30 GMT
Updated Branches:
  refs/heads/master abd3fa48a -> 2bd7053b0


FALCON-38 Falcon's parent workflow actions (pre-processing & post-processing) should have
multiple retries. Contributed by shaik.idris


Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/2bd7053b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/2bd7053b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/2bd7053b

Branch: refs/heads/master
Commit: 2bd7053b03f83a4c83383766728c3de76222ed89
Parents: abd3fa4
Author: shaikidris <psychidris@gmail.com>
Authored: Tue Jan 14 22:12:28 2014 +0530
Committer: shaikidris <psychidris@gmail.com>
Committed: Tue Jan 14 22:12:28 2014 +0530

----------------------------------------------------------------------
 CHANGES.txt                                     |  4 ++
 common/src/main/resources/runtime.properties    |  2 +
 .../falcon/converter/OozieFeedMapper.java       | 16 ++++++
 .../falcon/converter/OozieFeedMapperTest.java   | 22 +++++++-
 .../converter/AbstractOozieEntityMapper.java    | 13 +++++
 oozie/src/main/resources/oozie-workflow-0.3.xsd | 53 ++++++++++++++------
 .../falcon/converter/OozieProcessMapper.java    |  3 ++
 .../converter/OozieProcessMapperTest.java       |  6 +++
 8 files changed, 104 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/2bd7053b/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 6eac1ef..c5ceb7e 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -15,10 +15,14 @@ Trunk (Unreleased)
     
     FALCON-66 Make oozie version change configurable. (Shwetha GS
     via Srikanth Sundarrajan)
+    
+    FALCON-38 Falcon's parent workflow actions (pre-processing & prost-processing) should
have multiple retries. (Shaik Idris)
+
 
   OPTIMIZATIONS
 
   BUG FIXES
+
     FALCON-262 Example files should use aligned dependency versions. (Jean-Baptiste Onofré
     via Shwetha GS)
 

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/2bd7053b/common/src/main/resources/runtime.properties
----------------------------------------------------------------------
diff --git a/common/src/main/resources/runtime.properties b/common/src/main/resources/runtime.properties
index 26c916d..87b9d1e 100644
--- a/common/src/main/resources/runtime.properties
+++ b/common/src/main/resources/runtime.properties
@@ -23,3 +23,5 @@
 *.log.cleanup.frequency.days.retention =days(7)
 *.log.cleanup.frequency.months.retention =months(3)
 
+*.falcon.parentworkflow.retry.max=3
+*.falcon.parentworkflow.retry.interval.secs=1
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/2bd7053b/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
----------------------------------------------------------------------
diff --git a/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java b/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
index fb066ab..6ca2134 100644
--- a/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
+++ b/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
@@ -198,6 +198,7 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed>
{
                 WORKFLOWAPP retWfApp = getWorkflowTemplate(RETENTION_WF_TEMPLATE);
                 retWfApp.setName(wfName);
                 addLibExtensionsToWorkflow(cluster, retWfApp, EntityType.FEED, "retention");
+                addOozieRetries(retWfApp);
                 marshal(cluster, retWfApp, wfPath);
             } catch(IOException e) {
                 throw new FalconException("Unable to create retention workflow", e);
@@ -220,10 +221,12 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed>
{
                 WORKFLOWAPP repWFapp = getWorkflowTemplate(REPLICATION_WF_TEMPLATE);
                 repWFapp.setName(wfName);
                 addLibExtensionsToWorkflow(cluster, repWFapp, EntityType.FEED, "replication");
+                addOozieRetries(repWFapp);
                 marshal(cluster, repWFapp, wfPath);
             } catch(IOException e) {
                 throw new FalconException("Unable to create replication workflow", e);
             }
+
         }
 
         private COORDINATORAPP createAndGetCoord(Feed feed, Cluster srcCluster, Cluster trgCluster,
@@ -543,4 +546,17 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed>
{
             props.put(ARG.feedInstancePaths.getPropName(), instancePaths);
         }
     }
+
+    private void addOozieRetries(WORKFLOWAPP workflow) {
+        for (Object object : workflow.getDecisionOrForkOrJoin()) {
+            if (!(object instanceof org.apache.falcon.oozie.workflow.ACTION)) {
+                continue;
+            }
+            org.apache.falcon.oozie.workflow.ACTION action = (org.apache.falcon.oozie.workflow.ACTION)
object;
+            String actionName = action.getName();
+            if (FALCON_ACTIONS.contains(actionName)) {
+                decorateWithOozieRetries(action);
+            }
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/2bd7053b/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
----------------------------------------------------------------------
diff --git a/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java b/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
index de7f9e5..128784e 100644
--- a/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
+++ b/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
@@ -203,6 +203,23 @@ public class OozieFeedMapperTest {
         Assert.assertEquals(props.get("feedInstancePaths"), "${coord:dataIn('input')}");
 
         assertLibExtensions(coord, "replication");
+        assertWorkflowRetries(coord);
+    }
+
+    private void assertWorkflowRetries(COORDINATORAPP coord) throws JAXBException, IOException
{
+        WORKFLOWAPP wf = getWorkflowapp(coord);
+        List<Object> actions = wf.getDecisionOrForkOrJoin();
+        for (Object obj : actions) {
+            if (!(obj instanceof ACTION)) {
+                continue;
+            }
+            ACTION action = (ACTION) obj;
+            String actionName = action.getName();
+            if (AbstractOozieEntityMapper.FALCON_ACTIONS.contains(actionName)) {
+                Assert.assertEquals(action.getRetryMax(), "3");
+                Assert.assertEquals(action.getRetryInterval(), "1");
+            }
+        }
     }
 
     private void assertLibExtensions(COORDINATORAPP coord, String lifecycle) throws Exception
{
@@ -430,7 +447,7 @@ public class OozieFeedMapperTest {
     }
 
     @Test
-    public void testRetentionCoords() throws FalconException {
+    public void testRetentionCoords() throws FalconException, JAXBException, IOException
{
         org.apache.falcon.entity.v0.feed.Cluster cluster = FeedHelper.getCluster(feed, srcCluster.getName());
         final Calendar instance = Calendar.getInstance();
         instance.roll(Calendar.YEAR, 1);
@@ -468,5 +485,8 @@ public class OozieFeedMapperTest {
         // verify the post processing params
         Assert.assertEquals(props.get("feedNames"), feed.getName());
         Assert.assertEquals(props.get("feedInstancePaths"), "IGNORE");
+
+        assertWorkflowRetries(coord);
     }
+
 }

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/2bd7053b/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 ad095fd..cecdeef 100644
--- a/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
+++ b/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
@@ -39,6 +39,7 @@ import org.apache.falcon.oozie.workflow.ACTION;
 import org.apache.falcon.oozie.workflow.WORKFLOWAPP;
 import org.apache.falcon.service.FalconPathFilter;
 import org.apache.falcon.service.SharedLibraryHostingService;
+import org.apache.falcon.util.RuntimeProperties;
 import org.apache.falcon.util.StartupProperties;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
@@ -51,10 +52,14 @@ import org.apache.oozie.client.OozieClient;
 
 import javax.xml.bind.*;
 import java.io.*;
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Map.Entry;
+import java.util.Set;
 
 /**
  * Entity mapper base class that allows an entity to be mapped to oozie bundle.
@@ -75,6 +80,8 @@ public abstract class AbstractOozieEntityMapper<T extends Entity>
{
     protected static final JAXBContext COORD_JAXB_CONTEXT;
     protected static final JAXBContext BUNDLE_JAXB_CONTEXT;
     protected static final JAXBContext HIVE_ACTION_JAXB_CONTEXT;
+    public static final Set<String> FALCON_ACTIONS = new HashSet<String>(Arrays.asList(new
String[] { "recordsize",
+        "succeeded-post-processing", "failed-post-processing", "eviction", "jms-messaging",
}));
 
     protected static final FalconPathFilter FALCON_JAR_FILTER = new FalconPathFilter() {
         @Override
@@ -400,4 +407,10 @@ public abstract class AbstractOozieEntityMapper<T extends Entity>
{
             IOUtils.closeQuietly(out);
         }
     }
+
+    protected void decorateWithOozieRetries(ACTION action) {
+        Properties props = RuntimeProperties.get();
+        action.setRetryMax(props.getProperty("falcon.parentworkflow.retry.max", "3"));
+        action.setRetryInterval(props.getProperty("falcon.parentworkflow.retry.interval.secs",
"1"));
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/2bd7053b/oozie/src/main/resources/oozie-workflow-0.3.xsd
----------------------------------------------------------------------
diff --git a/oozie/src/main/resources/oozie-workflow-0.3.xsd b/oozie/src/main/resources/oozie-workflow-0.3.xsd
index 012d9f7..996a109 100644
--- a/oozie/src/main/resources/oozie-workflow-0.3.xsd
+++ b/oozie/src/main/resources/oozie-workflow-0.3.xsd
@@ -7,15 +7,15 @@
   to you under the Apache License, Version 2.0 (the
   "License"); you may not use this file except in compliance
   with the License.  You may obtain a copy of the License at
-
-      http://www.apache.org/licenses/LICENSE-2.0
-
+  
+       http://www.apache.org/licenses/LICENSE-2.0
+  
   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
-  -->
+-->
 <xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema" xmlns:workflow="uri:oozie:workflow:0.3"
            elementFormDefault="qualified" targetNamespace="uri:oozie:workflow:0.3">
 
@@ -29,6 +29,7 @@
 
     <xs:complexType name="WORKFLOW-APP">
         <xs:sequence>
+        	<xs:element name="credentials" type="workflow:CREDENTIALS" minOccurs="0" maxOccurs="1"/>
             <xs:element name="start" type="workflow:START" minOccurs="1" maxOccurs="1"/>
             <xs:choice minOccurs="0" maxOccurs="unbounded">
                 <xs:element name="decision" type="workflow:DECISION" minOccurs="1" maxOccurs="1"/>
@@ -40,7 +41,7 @@
             <xs:element name="end" type="workflow:END" minOccurs="1" maxOccurs="1"/>
             <xs:any namespace="uri:oozie:sla:0.1" minOccurs="0" maxOccurs="1"/>
         </xs:sequence>
-        <xs:attribute name="name" type="workflow:IDENTIFIER" use="required"/>
+        <xs:attribute name="name" type="xs:string" use="required"/>
     </xs:complexType>
 
     <xs:complexType name="START">
@@ -130,9 +131,12 @@
             </xs:choice>
             <xs:element name="ok" type="workflow:ACTION_TRANSITION" minOccurs="1" maxOccurs="1"/>
             <xs:element name="error" type="workflow:ACTION_TRANSITION" minOccurs="1" maxOccurs="1"/>
-            <!--<xs:any namespace="uri:oozie:sla:0.1" minOccurs="0" maxOccurs="1"/>-->
+            <!--xs:any namespace="uri:oozie:sla:0.1" minOccurs="0" maxOccurs="1"/-->
         </xs:sequence>
         <xs:attribute name="name" type="workflow:IDENTIFIER" use="required"/>
+        <xs:attribute name="cred" type="xs:string"/>
+        <xs:attribute name="retry-max" type="xs:string"/>
+        <xs:attribute name="retry-interval" type="xs:string"/>
     </xs:complexType>
 
     <xs:complexType name="MAP-REDUCE">
@@ -165,7 +169,7 @@
             <xs:element name="archive" type="xs:string" minOccurs="0" maxOccurs="unbounded"/>
         </xs:sequence>
     </xs:complexType>
-
+    
     <xs:complexType name="SSH">
         <xs:sequence>
             <xs:element name="host" type="xs:string" minOccurs="1" maxOccurs="1"/>
@@ -184,12 +188,12 @@
     </xs:complexType>
 
     <xs:complexType name="FS">
-        <xs:sequence>
-            <xs:element name="delete" type="workflow:DELETE" minOccurs="0" maxOccurs="unbounded"/>
-            <xs:element name="mkdir" type="workflow:MKDIR" minOccurs="0" maxOccurs="unbounded"/>
-            <xs:element name="move" type="workflow:MOVE" minOccurs="0" maxOccurs="unbounded"/>
-            <xs:element name="chmod" type="workflow:CHMOD" minOccurs="0" maxOccurs="unbounded"/>
-        </xs:sequence>
+	<xs:choice minOccurs="0" maxOccurs="unbounded">
+            <xs:element name="delete" type="workflow:DELETE"/>
+            <xs:element name="mkdir" type="workflow:MKDIR"/>
+            <xs:element name="move" type="workflow:MOVE"/>
+            <xs:element name="chmod" type="workflow:CHMOD"/>
+        </xs:choice>
     </xs:complexType>
 
     <xs:complexType name="JAVA">
@@ -270,5 +274,26 @@
         <xs:attribute name="permissions" type="xs:string" use="required"/>
         <xs:attribute name="dir-files" type="xs:string"/>
     </xs:complexType>
-
+    
+    <xs:complexType name="CREDENTIALS">             
+        <xs:sequence minOccurs="0" maxOccurs="unbounded">
+            <xs:element name="credential" type="workflow:CREDENTIAL"/>
+		</xs:sequence>                
+    </xs:complexType>
+    
+   	<xs:complexType name="CREDENTIAL">
+        <xs:sequence  minOccurs="0" maxOccurs="unbounded" >
+                   <xs:element name="property" minOccurs="1" maxOccurs="unbounded">
+                	<xs:complexType>
+	                    <xs:sequence>
+	                        <xs:element name="name" minOccurs="1" maxOccurs="1" type="xs:string"/>
+	                        <xs:element name="value" minOccurs="1" maxOccurs="1" type="xs:string"/>
+	                        <xs:element name="description" minOccurs="0" maxOccurs="1" type="xs:string"/>
+	                    </xs:sequence>
+                	</xs:complexType>
+           		</xs:element>
+        </xs:sequence>
+        <xs:attribute name="name" type="xs:string" use="required"/>
+        <xs:attribute name="type" type="xs:string" use="required"/>
+    </xs:complexType>
 </xs:schema>

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/2bd7053b/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java
----------------------------------------------------------------------
diff --git a/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java b/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java
index b72e243..87be709 100644
--- a/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java
+++ b/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java
@@ -519,6 +519,8 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process>
{
                 decoratePIGAction(cluster, process, processWorkflow, action.getPig(), parentWfPath);
             } else if (engineType == EngineType.HIVE && actionName.equals("user-hive-job"))
{
                 decorateHiveAction(cluster, process, processWorkflow, action, parentWfPath);
+            } else if (FALCON_ACTIONS.contains(actionName)) {
+                decorateWithOozieRetries(action);
             }
         }
 
@@ -812,4 +814,5 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process>
{
             throw new RuntimeException("Unable to marshall hive action.", e);
         }
     }
+
 }

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/2bd7053b/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
----------------------------------------------------------------------
diff --git a/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
b/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
index 224397e..794e585 100644
--- a/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
+++ b/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
@@ -434,6 +434,12 @@ public class OozieProcessMapperTest extends AbstractTestBase {
         Assert.assertEquals("user-oozie-workflow", ((ACTION) decisionOrForkOrJoin.get(5)).getName());
         Assert.assertEquals("succeeded-post-processing", ((ACTION) decisionOrForkOrJoin.get(6)).getName());
         Assert.assertEquals("failed-post-processing", ((ACTION) decisionOrForkOrJoin.get(7)).getName());
+        Assert.assertEquals("3", ((ACTION) decisionOrForkOrJoin.get(1)).getRetryMax());
+        Assert.assertEquals("1", ((ACTION) decisionOrForkOrJoin.get(1)).getRetryInterval());
+        Assert.assertEquals("3", ((ACTION) decisionOrForkOrJoin.get(6)).getRetryMax());
+        Assert.assertEquals("1", ((ACTION) decisionOrForkOrJoin.get(6)).getRetryInterval());
+        Assert.assertEquals("3", ((ACTION) decisionOrForkOrJoin.get(7)).getRetryMax());
+        Assert.assertEquals("1", ((ACTION) decisionOrForkOrJoin.get(7)).getRetryInterval());
     }
 
     private COORDINATORAPP getCoordinator(Path path) throws Exception {


Mime
View raw message