You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@falcon.apache.org by ve...@apache.org on 2013/09/24 01:04:56 UTC
[8/8] git commit: Merge remote-tracking branch 'origin/master' into 85
Merge remote-tracking branch 'origin/master' into 85
Conflicts:
CHANGES.txt
common/src/main/java/org/apache/falcon/entity/EntityUtil.java
common/src/main/java/org/apache/falcon/entity/parser/ClusterEntityParser.java
feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
pom.xml
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/d1c5eafe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/d1c5eafe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/d1c5eafe
Branch: refs/heads/FALCON-85
Commit: d1c5eafe08a1dde8b020b5f1f831e6d43138132c
Parents: 0879caf acb9f6a
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Mon Sep 23 16:04:14 2013 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Mon Sep 23 16:04:14 2013 -0700
----------------------------------------------------------------------
CHANGES.txt | 13 ++++
.../falcon/entity/ColoClusterRelation.java | 2 +-
.../org/apache/falcon/entity/EntityUtil.java | 8 ++
.../entity/parser/ClusterEntityParser.java | 7 +-
.../entity/parser/ProcessEntityParser.java | 10 +--
.../falcon/entity/store/ConfigurationStore.java | 49 ++++--------
.../apache/falcon/entity/v0/EntityGraph.java | 4 +-
.../org/apache/falcon/group/FeedGroupMap.java | 4 +-
.../service/ConfigurationChangeListener.java | 2 +-
common/src/main/resources/startup.properties | 3 +
.../entity/store/ConfigurationStoreTest.java | 44 ++++++++++-
docs/src/site/twiki/InstallationSteps.twiki | 37 ++++++---
.../falcon/converter/OozieFeedMapper.java | 63 ++++++++-------
.../falcon/converter/OozieFeedMapperTest.java | 62 ++++++++++++---
feed/src/test/resources/feed.xml | 4 +-
.../falcon/hadoop/FileSystemExtension.java | 28 +++++++
hadoop-webapp/pom.xml | 6 ++
hadoop-webapp/src/main/resources/core-site.xml | 5 ++
.../converter/AbstractOozieEntityMapper.java | 64 ++++++++++++---
.../service/SharedLibraryHostingService.java | 82 +++++++++++++-------
.../falcon/workflow/FalconPostProcessing.java | 7 +-
pom.xml | 4 +-
.../falcon/converter/OozieProcessMapper.java | 5 ++
.../converter/OozieProcessMapperTest.java | 53 ++++++++++---
rerun/pom.xml | 35 +++++++++
.../apache/falcon/retention/FeedEvictor.java | 9 ++-
.../falcon/retention/FeedEvictorTest.java | 16 ++--
src/bin/service-start.sh | 6 +-
webapp/pom.xml | 4 +
.../falcon/resource/EntityManagerJerseyIT.java | 41 ++++++++--
.../org/apache/falcon/resource/TestContext.java | 4 +-
webapp/src/test/resources/feed-template1.xml | 2 +-
32 files changed, 505 insertions(+), 178 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index e286b4e,b1fa477..b488cd0
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -5,23 -5,8 +5,25 @@@ Trunk (Unreleased
INCOMPATIBLE CHANGES
NEW FEATURES
+ FALCON-102 Add integration tests for feed entity parser for table.
+ (Venkatesh Seetharam)
+
+ FALCON-103 Upgrade oozie to 4.0.x. (Venkatesh Seetharam)
+
+ FALCON-96 Hive client to talk to the metastore. (Venkatesh
+ Seetharam via Srikanth Sundarrajan)
+
+ FALCON-87 Hive table integration with feed entity. (Venkatesh
+ Seetharam via Srikanth Sundarrajan)
+
+ FALCON-86 Hive table integration with cluster entity. (Venaktesh
+ Seetharam via Srikanth Sundarrajan)
+
+ FALCON-88 Add embedded hive and webhcat for integration tests.
+ (Venkatesh Seetharam via Srikanth Sundarrajan)
+
+ FALCON-107 Adding extensions. (Shwetha GS)
+
FALCON-62 Falcon compilation with hadoop 2.0 libs. (Shwetha GS
via Srikanth Sundarrajan)
@@@ -29,6 -14,9 +31,9 @@@
FALCON-131 Remove set -e from falcon scripts. (Suresh Srinivas
via Srikanth Sundarrajan)
- FALCON-69 Exception from ConfigurationChangeListener should fail
++ FALCON-69 Exception from ConfigurationChangeListener should fail
+ the API. (Shwetha GS via Srikanth Sundarrajan)
+
FALCON-63 Ability to ingest hadoop libs to falcon release package.
(Suhas Vasu via Shwetha GS)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/common/src/main/java/org/apache/falcon/entity/EntityUtil.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/common/src/main/java/org/apache/falcon/entity/parser/ClusterEntityParser.java
----------------------------------------------------------------------
diff --cc common/src/main/java/org/apache/falcon/entity/parser/ClusterEntityParser.java
index 631ec60,b1cf8f3..96fc6f4
--- a/common/src/main/java/org/apache/falcon/entity/parser/ClusterEntityParser.java
+++ b/common/src/main/java/org/apache/falcon/entity/parser/ClusterEntityParser.java
@@@ -23,14 -23,12 +23,14 @@@ import java.io.IOException
import javax.jms.ConnectionFactory;
import org.apache.falcon.FalconException;
+import org.apache.falcon.catalog.CatalogServiceFactory;
import org.apache.falcon.entity.ClusterHelper;
+ import org.apache.falcon.entity.EntityUtil;
import org.apache.falcon.entity.store.StoreAccessException;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.cluster.Interfacetype;
+import org.apache.falcon.entity.v0.cluster.Interface;
- import org.apache.falcon.util.DeploymentUtil;
import org.apache.falcon.util.StartupProperties;
import org.apache.falcon.workflow.WorkflowEngineFactory;
import org.apache.hadoop.conf.Configuration;
@@@ -58,14 -56,8 +58,11 @@@ public class ClusterEntityParser extend
validateScheme(cluster, Interfacetype.WRITE);
validateScheme(cluster, Interfacetype.WORKFLOW);
validateScheme(cluster, Interfacetype.MESSAGING);
+ if (ClusterHelper.getInterface(cluster, Interfacetype.REGISTRY) != null) {
+ validateScheme(cluster, Interfacetype.REGISTRY);
+ }
- // No interface validations in prism or other falcon servers.
- // Only the falcon server for which the cluster belongs to should validate interfaces
- if (DeploymentUtil.isPrism() || !cluster.getColo().equals(DeploymentUtil.getCurrentColo())) {
- LOG.info("No interface validations in prism or falcon servers not applicable.");
+ if (!EntityUtil.responsibleFor(cluster.getColo())) {
return;
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/common/src/main/java/org/apache/falcon/group/FeedGroupMap.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/common/src/main/resources/startup.properties
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/docs/src/site/twiki/InstallationSteps.twiki
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
----------------------------------------------------------------------
diff --cc feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
index 00c261b,b6052e7..c97ccf1
--- a/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
+++ b/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
@@@ -114,40 -114,16 +114,33 @@@ public class OozieFeedMapper extends Ab
Feed feed = getEntity();
ACTION retentionAction = new ACTION();
WORKFLOW retentionWorkflow = new WORKFLOW();
- try {
- //
- WORKFLOWAPP retWfApp = createRetentionWorkflow();
- retWfApp.setName(wfName);
- marshal(cluster, retWfApp, wfPath);
- retentionWorkflow.setAppPath(getStoragePath(wfPath.toString()));
-
- Map<String, String> props = createCoordDefaultConfiguration(cluster, wfPath, wfName);
+ createRetentionWorkflow(cluster, wfPath, wfName);
+ retentionWorkflow.setAppPath(getStoragePath(wfPath.toString()));
- org.apache.falcon.entity.v0.feed.Cluster feedCluster = FeedHelper.getCluster(feed, cluster.getName());
- String feedBasePaths = getFeedDataPath(cluster, feed);
+ Map<String, String> props = createCoordDefaultConfiguration(cluster, wfPath, wfName);
- props.put("feedDataPath", feedBasePaths);
- props.put("timeZone", feed.getTimezone().getID());
- props.put("frequency", feed.getFrequency().getTimeUnit().name());
- props.put("limit", feedCluster.getRetention().getLimit().toString());
- props.put(ARG.operation.getPropName(), EntityOps.DELETE.name());
- props.put(ARG.feedNames.getPropName(), feed.getName());
- props.put(ARG.feedInstancePaths.getPropName(), "IGNORE");
-
- retentionWorkflow.setConfiguration(getCoordConfig(props));
- retentionAction.setWorkflow(retentionWorkflow);
- return retentionAction;
- } catch (IOException e) {
- throw new FalconException("Unable to create parent/retention workflow", e);
- }
+ org.apache.falcon.entity.v0.feed.Cluster feedCluster = FeedHelper.getCluster(feed, cluster.getName());
- String feedPathMask = getLocationURI(cluster, feed, LocationType.DATA);
- String metaPathMask = getLocationURI(cluster, feed, LocationType.META);
- String statsPathMask = getLocationURI(cluster, feed, LocationType.STATS);
- String tmpPathMask = getLocationURI(cluster, feed, LocationType.TMP);
++ String feedBasePaths = getFeedDataPath(cluster, feed);
++
++ props.put("feedDataPath", feedBasePaths);
++ props.put("timeZone", feed.getTimezone().getID());
++ props.put("frequency", feed.getFrequency().getTimeUnit().name());
++ props.put("limit", feedCluster.getRetention().getLimit().toString());
++ props.put(ARG.operation.getPropName(), EntityOps.DELETE.name());
++ props.put(ARG.feedNames.getPropName(), feed.getName());
++ props.put(ARG.feedInstancePaths.getPropName(), "IGNORE");
++
++ retentionWorkflow.setConfiguration(getCoordConfig(props));
++ retentionAction.setWorkflow(retentionWorkflow);
++ return retentionAction;
+ }
+
+ protected String getFeedDataPath(Cluster cluster, Feed feed) throws FalconException {
+ final Storage storage = FeedHelper.createStorage(cluster, feed);
+ String feedPathMask = storage.getUriTemplate(LocationType.DATA);
+ String metaPathMask = storage.getUriTemplate(LocationType.META);
+ String statsPathMask = storage.getUriTemplate(LocationType.STATS);
+ String tmpPathMask = storage.getUriTemplate(LocationType.TMP);
StringBuilder feedBasePaths = new StringBuilder(feedPathMask);
if (metaPathMask != null) {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
----------------------------------------------------------------------
diff --cc feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
index 3b3132a,4fb6a58..cec278e
--- a/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
+++ b/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
@@@ -69,20 -73,28 +75,27 @@@ public class OozieFeedMapperTest
cleanupStore();
- srcCluster = (Cluster) storeEntity(EntityType.CLUSTER, SRC_CLUSTER_PATH);
- ClusterHelper.getInterface(srcCluster, Interfacetype.WRITE).setEndpoint(srcHdfsUrl);
+ srcCluster = (Cluster) storeEntity(EntityType.CLUSTER, SRC_CLUSTER_PATH, srcHdfsUrl);
- trgCluster = (Cluster) storeEntity(EntityType.CLUSTER, TRG_CLUSTER_PATH);
- ClusterHelper.getInterface(trgCluster, Interfacetype.WRITE).setEndpoint(trgHdfsUrl);
+ trgCluster = (Cluster) storeEntity(EntityType.CLUSTER, TRG_CLUSTER_PATH, trgHdfsUrl);
- feed = (Feed) storeEntity(EntityType.FEED, FEED);
+ feed = (Feed) storeEntity(EntityType.FEED, FEED, null);
-
}
- protected Entity storeEntity(EntityType type, String path) throws Exception {
+ protected Entity storeEntity(EntityType type, String template, String writeEndpoint) throws Exception {
Unmarshaller unmarshaller = type.getUnmarshaller();
Entity entity = (Entity) unmarshaller
- .unmarshal(OozieFeedMapperTest.class.getResource(path));
+ .unmarshal(OozieFeedMapperTest.class.getResource(template));
store.publish(type, entity);
+
+ if (type == EntityType.CLUSTER) {
+ Cluster cluster = (Cluster) entity;
+ ClusterHelper.getInterface(cluster, Interfacetype.WRITE).setEndpoint(writeEndpoint);
+ FileSystem fs = new Path(writeEndpoint).getFileSystem(new Configuration());
+ fs.create(new Path(ClusterHelper.getLocation(cluster, "working"), "libext/FEED/retention/ext.jar")).close();
+ fs.create(
+ new Path(ClusterHelper.getLocation(cluster, "working"), "libext/FEED/replication/ext.jar")).close();
+ }
return entity;
}
@@@ -149,35 -165,32 +166,62 @@@
break;
}
}
+ assertLibExtensions(coord, "replication");
+ }
+
+ private void assertLibExtensions(COORDINATORAPP coord, String lifecycle) throws Exception {
+ String wfPath = coord.getAction().getWorkflow().getAppPath().replace("${nameNode}", "");
+ JAXBContext jaxbContext = JAXBContext.newInstance(WORKFLOWAPP.class);
+ WORKFLOWAPP wf = ((JAXBElement<WORKFLOWAPP>) jaxbContext.createUnmarshaller().unmarshal(
+ trgMiniDFS.getFileSystem().open(new Path(wfPath, "workflow.xml")))).getValue();
+ List<Object> actions = wf.getDecisionOrForkOrJoin();
+ for (Object obj : actions) {
+ if (!(obj instanceof ACTION)) {
+ continue;
+ }
+ ACTION action = (ACTION) obj;
+ List<String> files = null;
+ if (action.getJava() != null) {
+ files = action.getJava().getFile();
+ } else if (action.getPig() != null) {
+ files = action.getPig().getFile();
+ } else if (action.getMapReduce() != null) {
+ files = action.getMapReduce().getFile();
+ }
+ if (files != null) {
+ Assert.assertTrue(files.get(files.size() - 1).endsWith("/projects/falcon/working/libext/FEED/"
+ + lifecycle + "/ext.jar"));
+ }
+ }
}
+
+ @Test
+ public void testRetentionCoords() throws FalconException {
+ org.apache.falcon.entity.v0.feed.Cluster cluster = FeedHelper.getCluster(feed, srcCluster.getName());
+ final Calendar instance = Calendar.getInstance();
+ instance.roll(Calendar.YEAR, 1);
+ cluster.getValidity().setEnd(instance.getTime());
+
+ OozieFeedMapper feedMapper = new OozieFeedMapper(feed);
+ List<COORDINATORAPP> coords = feedMapper.getCoordinators(srcCluster, new Path("/projects/falcon/"));
+ COORDINATORAPP coord = coords.get(0);
+
+ Assert.assertEquals(coord.getAction().getWorkflow().getAppPath(), "${nameNode}/projects/falcon/RETENTION");
+ Assert.assertEquals(coord.getName(), "FALCON_FEED_RETENTION_" + feed.getName());
+ Assert.assertEquals(coord.getFrequency(), "${coord:hours(6)}");
+
+ String feedDataPath = null;
+ org.apache.falcon.oozie.coordinator.CONFIGURATION configuration =
+ coord.getAction().getWorkflow().getConfiguration();
+ for (Property property : configuration.getProperty()) {
+ if ("feedDataPath".equals(property.getName())) {
+ feedDataPath = property.getValue();
+ break;
+ }
+ }
+
+ if (feedDataPath != null) {
+ Assert.assertEquals(feedDataPath, feedMapper.getFeedDataPath(srcCluster, feed));
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/hadoop-webapp/pom.xml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/pom.xml
----------------------------------------------------------------------
diff --cc pom.xml
index 6dc34e3,1a47e19..6eda8f5
--- a/pom.xml
+++ b/pom.xml
@@@ -98,13 -98,10 +98,13 @@@
<hadoop.profile>1</hadoop.profile>
<hadoop.version>1.1.2</hadoop.version>
- <slf4j.version>1.2</slf4j.version>
- <oozie.version>3.2.2</oozie.version>
+ <slf4j.version>1.6.1</slf4j.version>
+ <oozie.version>4.0.0</oozie.version>
+ <falcon.oozie.version>${oozie.version}-falcon</falcon.oozie.version>
<activemq.version>5.4.3</activemq.version>
- <hadoop-distcp.version>0.3</hadoop-distcp.version>
+ <hive.version>0.11.0</hive.version>
+ <hcatalog.version>0.11.0</hcatalog.version>
+ <hadoop-distcp.version>0.9</hadoop-distcp.version>
<jetty.version>6.1.26</jetty.version>
<internal.maven.repo>file:///tmp/falcontemprepo</internal.maven.repo>
<skipCheck>false</skipCheck>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d1c5eafe/webapp/pom.xml
----------------------------------------------------------------------