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 2014/10/09 01:44:40 UTC
[1/5] git commit: FALCON-788 Lineage: Minor bug fixes. Contributed by
Sowmya Ramesh
Repository: incubator-falcon
Updated Branches:
refs/heads/master 5c305434c -> 396db7825
FALCON-788 Lineage: Minor bug fixes. Contributed by Sowmya Ramesh
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/b0929b44
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/b0929b44
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/b0929b44
Branch: refs/heads/master
Commit: b0929b4432870d1ffda79604b2ed29596934cc7a
Parents: 152afb1
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Wed Oct 8 14:56:36 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Wed Oct 8 16:20:38 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../metadata/AbstractMetadataResource.java | 20 ++++++++-
.../metadata/LineageMetadataResource.java | 43 ++++----------------
.../metadata/MetadataDiscoveryResource.java | 7 ++--
.../metadata/LineageMetadataResourceTest.java | 4 +-
5 files changed, 33 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b0929b44/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index b17afa3..dca9fa9 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -112,6 +112,8 @@ Trunk (Unreleased)
OPTIMIZATIONS
BUG FIXES
+ FALCON-788 Lineage: Minor bug fixes (Sowmya Ramesh via Venkatesh Seetharam)
+
FALCON-777 UT intermittent failures in Messaging tests depending on Thread.sleep
(Sowmya Ramesh via Venkatesh Seetharam)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b0929b44/prism/src/main/java/org/apache/falcon/resource/metadata/AbstractMetadataResource.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/resource/metadata/AbstractMetadataResource.java b/prism/src/main/java/org/apache/falcon/resource/metadata/AbstractMetadataResource.java
index a6c9b78..e9c90fc 100644
--- a/prism/src/main/java/org/apache/falcon/resource/metadata/AbstractMetadataResource.java
+++ b/prism/src/main/java/org/apache/falcon/resource/metadata/AbstractMetadataResource.java
@@ -25,6 +25,9 @@ import org.apache.falcon.service.Services;
import java.util.Set;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Response;
+
/**
* A base class for managing Metadata operations.
]*/
@@ -35,23 +38,36 @@ public abstract class AbstractMetadataResource {
protected MetadataMappingService service;
public AbstractMetadataResource() {
- service = (MetadataMappingService)getService(MetadataMappingService.SERVICE_NAME);
+ FalconService falconService = getService(MetadataMappingService.SERVICE_NAME);
+ service = (falconService != null) ? (MetadataMappingService)falconService : null;
}
- public static FalconService getService(String serviceName) {
+ private static FalconService getService(String serviceName) {
return (Services.get().isRegistered(serviceName))
? Services.get().getService(serviceName) : null;
}
protected Graph getGraph() {
+ checkIfMetadataMappingServiceIsEnabled();
return service.getGraph();
}
protected Set<String> getVertexIndexedKeys() {
+ checkIfMetadataMappingServiceIsEnabled();
return service.getVertexIndexedKeys();
}
protected Set<String> getEdgeIndexedKeys() {
+ checkIfMetadataMappingServiceIsEnabled();
return service.getEdgeIndexedKeys();
}
+
+ private void checkIfMetadataMappingServiceIsEnabled() {
+ if (service == null) {
+ throw new WebApplicationException(Response.status(Response.Status.NOT_FOUND)
+ .entity("Lineage " + MetadataMappingService.SERVICE_NAME + " is not enabled.")
+ .type("text/plain")
+ .build());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b0929b44/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java b/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
index 2181bea..ed50ce0 100644
--- a/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
+++ b/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
@@ -26,6 +26,7 @@ import com.tinkerpop.blueprints.VertexQuery;
import com.tinkerpop.blueprints.util.io.graphson.GraphSONMode;
import com.tinkerpop.blueprints.util.io.graphson.GraphSONUtility;
import org.apache.commons.lang.StringUtils;
+import org.apache.falcon.FalconWebException;
import org.apache.falcon.metadata.GraphUtils;
import org.apache.falcon.metadata.RelationshipLabel;
import org.apache.falcon.metadata.RelationshipProperty;
@@ -69,7 +70,6 @@ public class LineageMetadataResource extends AbstractMetadataResource {
@Path("/serialize")
@Produces({MediaType.APPLICATION_JSON})
public Response serializeGraph() {
- checkIfMetadataMappingServiceIsEnabled();
String file = StartupProperties.get().getProperty("falcon.graph.serialize.path")
+ "/lineage-graph-" + System.currentTimeMillis() + ".json";
LOG.info("Serialize Graph to: {}", file);
@@ -77,8 +77,7 @@ public class LineageMetadataResource extends AbstractMetadataResource {
GraphUtils.dump(getGraph(), file);
return Response.ok().build();
} catch (Exception e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
}
@@ -92,14 +91,12 @@ public class LineageMetadataResource extends AbstractMetadataResource {
@Path("/vertices/all")
@Produces({MediaType.APPLICATION_JSON})
public Response getVertices() {
- checkIfMetadataMappingServiceIsEnabled();
LOG.info("Get All Vertices");
try {
JSONObject response = buildJSONResponse(getGraph().getVertices());
return Response.ok(response).build();
} catch (JSONException e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
}
@@ -113,7 +110,6 @@ public class LineageMetadataResource extends AbstractMetadataResource {
@Path("/vertices/{id}")
@Produces({MediaType.APPLICATION_JSON})
public Response getVertex(@PathParam("id") final String vertexId) {
- checkIfMetadataMappingServiceIsEnabled();
LOG.info("Get vertex for vertexId= {}", vertexId);
validateInputs("Invalid argument: vertex id passed is null or empty.", vertexId);
try {
@@ -124,8 +120,7 @@ public class LineageMetadataResource extends AbstractMetadataResource {
vertex, getVertexIndexedKeys(), GraphSONMode.NORMAL));
return Response.ok(response).build();
} catch (JSONException e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
}
@@ -153,7 +148,6 @@ public class LineageMetadataResource extends AbstractMetadataResource {
public Response getVertexProperties(@PathParam("id") final String vertexId,
@DefaultValue("false") @QueryParam("relationships")
final String relationships) {
- checkIfMetadataMappingServiceIsEnabled();
LOG.info("Get vertex for vertexId= {}", vertexId);
validateInputs("Invalid argument: vertex id passed is null or empty.", vertexId);
try {
@@ -166,8 +160,7 @@ public class LineageMetadataResource extends AbstractMetadataResource {
response.put(TOTAL_SIZE, vertexProperties.size());
return Response.ok(response).build();
} catch (JSONException e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
}
@@ -253,7 +246,6 @@ public class LineageMetadataResource extends AbstractMetadataResource {
@Produces({MediaType.APPLICATION_JSON})
public Response getVertices(@QueryParam("key") final String key,
@QueryParam("value") final String value) {
- checkIfMetadataMappingServiceIsEnabled();
LOG.info("Get vertices for property key= {}, value= {}", key, value);
validateInputs("Invalid argument: key or value passed is null or empty.", key, value);
try {
@@ -261,8 +253,7 @@ public class LineageMetadataResource extends AbstractMetadataResource {
return Response.ok(response).build();
} catch (JSONException e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
}
@@ -278,7 +269,6 @@ public class LineageMetadataResource extends AbstractMetadataResource {
@Produces({MediaType.APPLICATION_JSON})
public Response getVertexEdges(@PathParam("id") String vertexId,
@PathParam("direction") String direction) {
- checkIfMetadataMappingServiceIsEnabled();
LOG.info("Get vertex edges for vertexId= {}, direction= {}", vertexId, direction);
// Validate vertex id. Direction is validated in VertexQueryArguments.
validateInputs("Invalid argument: vertex id or direction passed is null or empty.", vertexId, direction);
@@ -288,8 +278,7 @@ public class LineageMetadataResource extends AbstractMetadataResource {
return getVertexEdges(vertex, direction);
} catch (JSONException e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
}
@@ -347,15 +336,13 @@ public class LineageMetadataResource extends AbstractMetadataResource {
@Path("/edges/all")
@Produces({MediaType.APPLICATION_JSON})
public Response getEdges() {
- checkIfMetadataMappingServiceIsEnabled();
LOG.info("Get All Edges.");
try {
JSONObject response = buildJSONResponse(getGraph().getEdges());
return Response.ok(response).build();
} catch (JSONException e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
}
@@ -369,7 +356,6 @@ public class LineageMetadataResource extends AbstractMetadataResource {
@Path("/edges/{id}")
@Produces({MediaType.APPLICATION_JSON})
public Response getEdge(@PathParam("id") final String edgeId) {
- checkIfMetadataMappingServiceIsEnabled();
LOG.info("Get vertex for edgeId= {}", edgeId);
validateInputs("Invalid argument: edge id passed is null or empty.", edgeId);
try {
@@ -386,8 +372,7 @@ public class LineageMetadataResource extends AbstractMetadataResource {
edge, getEdgeIndexedKeys(), GraphSONMode.NORMAL));
return Response.ok(response).build();
} catch (JSONException e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
}
@@ -407,16 +392,6 @@ public class LineageMetadataResource extends AbstractMetadataResource {
return response;
}
- private void checkIfMetadataMappingServiceIsEnabled() {
- if (service == null) {
- throw new WebApplicationException(
- Response.status(Response.Status.NOT_FOUND)
- .entity("Lineage Metadata Service is not enabled.")
- .type("text/plain")
- .build());
- }
- }
-
private static void validateInputs(String errorMsg, String... inputs) {
for (String input : inputs) {
if (StringUtils.isEmpty(input)) {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b0929b44/prism/src/main/java/org/apache/falcon/resource/metadata/MetadataDiscoveryResource.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/resource/metadata/MetadataDiscoveryResource.java b/prism/src/main/java/org/apache/falcon/resource/metadata/MetadataDiscoveryResource.java
index 63ed3b6..1983371 100644
--- a/prism/src/main/java/org/apache/falcon/resource/metadata/MetadataDiscoveryResource.java
+++ b/prism/src/main/java/org/apache/falcon/resource/metadata/MetadataDiscoveryResource.java
@@ -25,6 +25,7 @@ import com.tinkerpop.blueprints.Vertex;
import com.tinkerpop.blueprints.util.io.graphson.GraphSONMode;
import com.tinkerpop.blueprints.util.io.graphson.GraphSONUtility;
import org.apache.commons.lang.StringUtils;
+import org.apache.falcon.FalconWebException;
import org.apache.falcon.metadata.RelationshipLabel;
import org.apache.falcon.metadata.RelationshipProperty;
import org.apache.falcon.metadata.RelationshipType;
@@ -81,8 +82,7 @@ public class MetadataDiscoveryResource extends AbstractMetadataResource {
response.put(TOTAL_SIZE, dimensionValues.length());
return Response.ok(response).build();
} catch (JSONException e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
}
@@ -117,8 +117,7 @@ public class MetadataDiscoveryResource extends AbstractMetadataResource {
vertexProperties.put("outVertices", getAdjacentVerticesJson(outEdges, Direction.IN));
} catch (JSONException e) {
- throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
- .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ throw FalconWebException.newException(e, Response.Status.INTERNAL_SERVER_ERROR);
}
return Response.ok(vertexProperties).build();
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b0929b44/prism/src/test/java/org/apache/falcon/resource/metadata/LineageMetadataResourceTest.java
----------------------------------------------------------------------
diff --git a/prism/src/test/java/org/apache/falcon/resource/metadata/LineageMetadataResourceTest.java b/prism/src/test/java/org/apache/falcon/resource/metadata/LineageMetadataResourceTest.java
index 0a8efc9..cabb44c 100644
--- a/prism/src/test/java/org/apache/falcon/resource/metadata/LineageMetadataResourceTest.java
+++ b/prism/src/test/java/org/apache/falcon/resource/metadata/LineageMetadataResourceTest.java
@@ -344,9 +344,7 @@ public class LineageMetadataResourceTest {
Services.get().reset();
try {
LineageMetadataResource resource = new LineageMetadataResource();
- Response response = resource.getVertices();
- Assert.assertEquals(response.getStatus(), Response.Status.NOT_FOUND.getStatusCode());
- Assert.assertEquals(response.getEntity().toString(), "Lineage Metadata Service is not enabled.");
+ resource.getVertices();
} finally {
Services.get().register(new WorkflowJobEndNotificationService());
Services.get().register(testContext.getService());
[4/5] git commit: FALCON-789 Post processing is broken for Feeds.
Contributed by Sowmya Ramesh
Posted by ve...@apache.org.
FALCON-789 Post processing is broken for Feeds. Contributed by Sowmya Ramesh
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/43110bd9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/43110bd9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/43110bd9
Branch: refs/heads/master
Commit: 43110bd9d5708963a826e9c39389ec99491a5355
Parents: d82c01d
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Wed Oct 8 16:24:03 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Wed Oct 8 16:24:03 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 3 +++
.../workflow/WorkflowExecutionContext.java | 24 ++++++++++-------
.../workflow/WorkflowExecutionContextTest.java | 28 +++++++++++++++++---
.../messaging/JMSMessageConsumerTest.java | 6 +++--
4 files changed, 46 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/43110bd9/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index c5ba06c..d7571ba 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -115,6 +115,9 @@ Trunk (Unreleased)
OPTIMIZATIONS
BUG FIXES
+ FALCON-789 Post processing is broken for Feeds (Sowmya Ramesh via
+ Venkatesh Seetharam)
+
FALCON-788 Lineage: Minor bug fixes (Sowmya Ramesh via Venkatesh Seetharam)
FALCON-777 UT intermittent failures in Messaging tests depending on Thread.sleep
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/43110bd9/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionContext.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionContext.java b/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionContext.java
index 8df6855..f3e643e 100644
--- a/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionContext.java
+++ b/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionContext.java
@@ -126,9 +126,7 @@ public class WorkflowExecutionContext {
}
public String getContextFile() {
- return EntityType.PROCESS.name().equals(getEntityType())
- ? getValue(WorkflowExecutionArgs.CONTEXT_FILE)
- : "/context/" + getValue(WorkflowExecutionArgs.CONTEXT_FILE); // needed by feed clean up
+ return getValue(WorkflowExecutionArgs.CONTEXT_FILE);
}
public String getLogDir() {
@@ -326,9 +324,14 @@ public class WorkflowExecutionContext {
}
}
- public static String getFilePath(String logDir, String entityName) {
+ public static String getFilePath(String logDir, String entityName, String entityType,
+ EntityOperations operation) {
+ // needed by feed clean up
+ String parentSuffix = EntityType.PROCESS.name().equals(entityType)
+ || EntityOperations.REPLICATE == operation ? "" : "/context/";
+
// LOG_DIR is sufficiently unique
- return new Path(logDir, entityName + "-wf-post-exec-context.json").toString();
+ return new Path(logDir + parentSuffix, entityName + "-wf-post-exec-context.json").toString();
}
public static WorkflowExecutionContext create(String[] args, Type type) throws FalconException {
@@ -346,12 +349,13 @@ public class WorkflowExecutionContext {
throw new FalconException("Error parsing wf args", e);
}
- wfProperties.put(WorkflowExecutionArgs.CONTEXT_TYPE, type.name());
- wfProperties.put(WorkflowExecutionArgs.CONTEXT_FILE,
- getFilePath(wfProperties.get(WorkflowExecutionArgs.LOG_DIR),
- wfProperties.get(WorkflowExecutionArgs.ENTITY_NAME)));
+ WorkflowExecutionContext executionContext = new WorkflowExecutionContext(wfProperties);
+ executionContext.context.put(WorkflowExecutionArgs.CONTEXT_TYPE, type.name());
+ executionContext.context.put(WorkflowExecutionArgs.CONTEXT_FILE,
+ getFilePath(executionContext.getLogDir(), executionContext.getEntityName(),
+ executionContext.getEntityType(), executionContext.getOperation()));
- return new WorkflowExecutionContext(wfProperties);
+ return executionContext;
}
private static CommandLine getCommand(String[] arguments) throws ParseException {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/43110bd9/common/src/test/java/org/apache/falcon/workflow/WorkflowExecutionContextTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/workflow/WorkflowExecutionContextTest.java b/common/src/test/java/org/apache/falcon/workflow/WorkflowExecutionContextTest.java
index 117f2b3..65a057d 100644
--- a/common/src/test/java/org/apache/falcon/workflow/WorkflowExecutionContextTest.java
+++ b/common/src/test/java/org/apache/falcon/workflow/WorkflowExecutionContextTest.java
@@ -96,7 +96,8 @@ public class WorkflowExecutionContextTest {
@Test
public void testGetContextFile() throws Exception {
Assert.assertEquals(context.getContextFile(),
- WorkflowExecutionContext.getFilePath(context.getLogDir(), context.getEntityName()));
+ WorkflowExecutionContext.getFilePath(context.getLogDir(), context.getEntityName(),
+ context.getEntityType(), context.getOperation()));
}
@Test
@@ -257,9 +258,30 @@ public class WorkflowExecutionContextTest {
}
@Test
- public void testGetFilePath() throws Exception {
- Assert.assertEquals(WorkflowExecutionContext.getFilePath(LOGS_DIR, ENTITY_NAME),
+ public void testGetFilePathForProcess() throws Exception {
+ final String filePath = WorkflowExecutionContext.getFilePath(LOGS_DIR,
+ ENTITY_NAME, "PROCESS", WorkflowExecutionContext.EntityOperations.GENERATE);
+ Assert.assertEquals(filePath,
LOGS_DIR + "/" + ENTITY_NAME + "-wf-post-exec-context.json");
+ Assert.assertEquals(context.getContextFile(), filePath);
+ }
+
+
+ @Test
+ public void testGetFilePathForFeedRetention() throws Exception {
+ final String filePath = WorkflowExecutionContext.getFilePath(LOGS_DIR,
+ ENTITY_NAME, "FEED", WorkflowExecutionContext.EntityOperations.DELETE);
+ Assert.assertEquals(filePath,
+ LOGS_DIR + "/context/" + ENTITY_NAME + "-wf-post-exec-context.json");
+ }
+
+ @Test
+ public void testGetFilePathForFeedReplication() throws Exception {
+ final String filePath = WorkflowExecutionContext.getFilePath(LOGS_DIR,
+ ENTITY_NAME, "FEED", WorkflowExecutionContext.EntityOperations.REPLICATE);
+ Assert.assertEquals(filePath,
+ LOGS_DIR + "/" + ENTITY_NAME + "-wf-post-exec-context.json");
+ Assert.assertEquals(context.getContextFile(), filePath);
}
private static String[] getTestMessageArgs() {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/43110bd9/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageConsumerTest.java
----------------------------------------------------------------------
diff --git a/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageConsumerTest.java b/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageConsumerTest.java
index 974116d..fa9ecbf 100644
--- a/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageConsumerTest.java
+++ b/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageConsumerTest.java
@@ -76,7 +76,8 @@ public class JMSMessageConsumerTest {
for (int i = 0; i < 3; i++) {
WorkflowExecutionContext context = WorkflowExecutionContext.create(
getMockFalconMessage(i), WorkflowExecutionContext.Type.POST_PROCESSING);
- context.serialize(WorkflowExecutionContext.getFilePath("/tmp/log", "process1"));
+ context.serialize(WorkflowExecutionContext.getFilePath("/tmp/log", "process1",
+ "process", WorkflowExecutionContext.EntityOperations.GENERATE));
MapMessage message = session.createMapMessage();
for (Map.Entry<WorkflowExecutionArgs, String> entry : context.entrySet()) {
@@ -89,7 +90,8 @@ public class JMSMessageConsumerTest {
WorkflowExecutionContext context = WorkflowExecutionContext.create(
getMockFalconMessage(5), WorkflowExecutionContext.Type.POST_PROCESSING);
- context.serialize(WorkflowExecutionContext.getFilePath("/tmp/log", "process1"));
+ context.serialize(WorkflowExecutionContext.getFilePath("/tmp/log", "process1",
+ "process", WorkflowExecutionContext.EntityOperations.GENERATE));
MapMessage mapMessage = session.createMapMessage();
for (Map.Entry<WorkflowExecutionArgs, String> entry : context.entrySet()) {
[2/5] git commit: FALCON-777 UT intermittent failures in Messaging
tests depending on Thread.sleep. Contributed by Sowmya Ramesh
Posted by ve...@apache.org.
FALCON-777 UT intermittent failures in Messaging tests depending on Thread.sleep. Contributed by Sowmya Ramesh
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/152afb1c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/152afb1c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/152afb1c
Branch: refs/heads/master
Commit: 152afb1c4a858f9ca4308016f02ef5037e3e756a
Parents: 5c30543
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Wed Oct 8 14:54:40 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Wed Oct 8 16:20:38 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 3 +
.../falcon/messaging/FeedProducerTest.java | 61 ++++++++------------
.../messaging/JMSMessageProducerTest.java | 7 ++-
.../falcon/messaging/ProcessProducerTest.java | 40 ++++++-------
.../workflow/FalconPostProcessingTest.java | 50 ++++++++++------
5 files changed, 83 insertions(+), 78 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/152afb1c/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index c76b6f0..b17afa3 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -112,6 +112,9 @@ Trunk (Unreleased)
OPTIMIZATIONS
BUG FIXES
+ FALCON-777 UT intermittent failures in Messaging tests depending on Thread.sleep
+ (Sowmya Ramesh via Venkatesh Seetharam)
+
FALCON-773 Log clean up handlers only work in distributed mode
(Balu Vellanki via Venkatesh Seetharam)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/152afb1c/messaging/src/test/java/org/apache/falcon/messaging/FeedProducerTest.java
----------------------------------------------------------------------
diff --git a/messaging/src/test/java/org/apache/falcon/messaging/FeedProducerTest.java b/messaging/src/test/java/org/apache/falcon/messaging/FeedProducerTest.java
index 6c8a926..2f5aa70 100644
--- a/messaging/src/test/java/org/apache/falcon/messaging/FeedProducerTest.java
+++ b/messaging/src/test/java/org/apache/falcon/messaging/FeedProducerTest.java
@@ -20,6 +20,7 @@ package org.apache.falcon.messaging;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.concurrent.CountDownLatch;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
@@ -40,6 +41,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
+import org.apache.commons.lang.StringUtils;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -61,6 +63,11 @@ public class FeedProducerTest {
private volatile AssertionError error;
private EmbeddedCluster dfsCluster;
private Configuration conf;
+ private CountDownLatch latch = new CountDownLatch(1);
+ private String[] instancePaths = {"/falcon/feed/agg-logs/path1/2010/10/10/20",
+ "/falcon/feed/agg-logs/path1/2010/10/10/21",
+ "/falcon/feed/agg-logs/path1/2010/10/10/22",
+ "/falcon/feed/agg-logs/path1/2010/10/10/23", };
@BeforeClass
public void setup() throws Exception {
@@ -111,12 +118,7 @@ public class FeedProducerTest {
public void testLogFile() throws Exception {
FileSystem fs = dfsCluster.getFileSystem();
OutputStream out = fs.create(logFile);
- InputStream in = new ByteArrayInputStream(
- ("instancePaths=/falcon/feed/agg-logs/path1/2010/10/10/20,"
- + "/falcon/feed/agg-logs/path1/2010/10/10/21,"
- + "/falcon/feed/agg-logs/path1/2010/10/10/22,"
- + "/falcon/feed/agg-logs/path1/2010/10/10/23")
- .getBytes());
+ InputStream in = new ByteArrayInputStream(("instancePaths=" + StringUtils.join(instancePaths, ",")).getBytes());
IOUtils.copyBytes(in, out, conf);
testProcessMessageCreator();
}
@@ -150,8 +152,9 @@ public class FeedProducerTest {
}
};
t.start();
- Thread.sleep(100);
+ // Wait for consumer to be ready
+ latch.await();
WorkflowExecutionContext context = WorkflowExecutionContext.create(
args, WorkflowExecutionContext.Type.POST_PROCESSING);
JMSMessageProducer jmsMessageProducer = JMSMessageProducer.builder(context)
@@ -173,41 +176,23 @@ public class FeedProducerTest {
Destination destination = session.createTopic(TOPIC_NAME);
MessageConsumer consumer = session.createConsumer(destination);
- // wait till you get atleast one message
- MapMessage m;
- for (m = null; m == null;) {
- m = (MapMessage) consumer.receive();
- }
- System.out.println("Consumed: " + m.toString());
- assertMessage(m);
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
- "/falcon/feed/agg-logs/path1/2010/10/10/20");
+ latch.countDown();
+ verifyMesssage(consumer);
- for (m = null; m == null;) {
- m = (MapMessage) consumer.receive();
- }
- System.out.println("Consumed: " + m.toString());
- assertMessage(m);
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
- "/falcon/feed/agg-logs/path1/2010/10/10/21");
+ connection.close();
+ }
- for (m = null; m == null;) {
- m = (MapMessage) consumer.receive();
- }
- System.out.println("Consumed: " + m.toString());
- assertMessage(m);
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
- "/falcon/feed/agg-logs/path1/2010/10/10/22");
+ private void verifyMesssage(MessageConsumer consumer) throws JMSException {
+ for (String instancePath : instancePaths) {
+ // receive call is blocking
+ MapMessage m = (MapMessage) consumer.receive();
- for (m = null; m == null;) {
- m = (MapMessage) consumer.receive();
+ System.out.println("Received JMS message {}" + m.toString());
+ System.out.println("Consumed: " + m.toString());
+ assertMessage(m);
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
+ instancePath);
}
- System.out.println("Consumed: " + m.toString());
- assertMessage(m);
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
- "/falcon/feed/agg-logs/path1/2010/10/10/23");
-
- connection.close();
}
private void assertMessage(MapMessage m) throws JMSException {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/152afb1c/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageProducerTest.java
----------------------------------------------------------------------
diff --git a/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageProducerTest.java b/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageProducerTest.java
index 490292f..d4373de 100644
--- a/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageProducerTest.java
+++ b/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageProducerTest.java
@@ -37,6 +37,7 @@ import javax.jms.Session;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
/**
* Test for falcon topic message producer.
@@ -51,6 +52,7 @@ public class JMSMessageProducerTest {
private List<MapMessage> mapMessages;
private volatile AssertionError error;
+ private CountDownLatch latch = new CountDownLatch(1);
@BeforeClass
public void setup() throws Exception {
@@ -175,7 +177,9 @@ public class JMSMessageProducerTest {
}
};
t.start();
- Thread.sleep(100);
+
+ // Wait for consumer to be ready
+ latch.await();
for (String[] message : messages) {
WorkflowExecutionContext context = WorkflowExecutionContext.create(
@@ -200,6 +204,7 @@ public class JMSMessageProducerTest {
Destination destination = session.createTopic(topicsToListen);
MessageConsumer consumer = session.createConsumer(destination);
+ latch.countDown();
mapMessages = new ArrayList<MapMessage>();
for (int i=0; i<size; i++) {
MapMessage m = (MapMessage) consumer.receive();
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/152afb1c/messaging/src/test/java/org/apache/falcon/messaging/ProcessProducerTest.java
----------------------------------------------------------------------
diff --git a/messaging/src/test/java/org/apache/falcon/messaging/ProcessProducerTest.java b/messaging/src/test/java/org/apache/falcon/messaging/ProcessProducerTest.java
index 80c2701..1cd3310 100644
--- a/messaging/src/test/java/org/apache/falcon/messaging/ProcessProducerTest.java
+++ b/messaging/src/test/java/org/apache/falcon/messaging/ProcessProducerTest.java
@@ -22,6 +22,7 @@ import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.falcon.workflow.WorkflowExecutionArgs;
import org.apache.falcon.workflow.WorkflowExecutionContext;
+import org.apache.commons.lang.StringUtils;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -34,6 +35,7 @@ import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageConsumer;
import javax.jms.Session;
+import java.util.concurrent.CountDownLatch;
/**
* Test for process message producer.
@@ -47,14 +49,16 @@ public class ProcessProducerTest {
private BrokerService broker;
private volatile AssertionError error;
+ private CountDownLatch latch = new CountDownLatch(1);
+ private String[] outputFeedNames = {"click-logs", "raw-logs"};
+ private String[] outputFeedPaths = {"/click-logs/10/05/05/00/20", "/raw-logs/10/05/05/00/20"};
@BeforeClass
public void setup() throws Exception {
args = new String[] {
"-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), ENTITY_NAME,
- "-" + WorkflowExecutionArgs.OUTPUT_FEED_NAMES.getName(), "click-logs,raw-logs",
- "-" + WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName(),
- "/click-logs/10/05/05/00/20,/raw-logs/10/05/05/00/20",
+ "-" + WorkflowExecutionArgs.OUTPUT_FEED_NAMES.getName(), StringUtils.join(outputFeedNames, ","),
+ "-" + WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName(), StringUtils.join(outputFeedPaths, ","),
"-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
"-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), "falcon",
"-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
@@ -103,7 +107,7 @@ public class ProcessProducerTest {
}
};
t.start();
- Thread.sleep(100);
+ latch.await();
WorkflowExecutionContext context = WorkflowExecutionContext.create(
args, WorkflowExecutionContext.Type.POST_PROCESSING);
@@ -126,25 +130,17 @@ public class ProcessProducerTest {
Destination destination = session.createTopic(getTopicName());
MessageConsumer consumer = session.createConsumer(destination);
- // wait till you get at least one message
- MapMessage m;
- for (m = null; m == null;) {
- m = (MapMessage) consumer.receive();
- }
- System.out.println("Consumed: " + m.toString());
- assertMessage(m);
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_NAMES.getName()), "click-logs");
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
- "/click-logs/10/05/05/00/20");
-
- for (m = null; m == null;) {
- m = (MapMessage) consumer.receive();
+ latch.countDown();
+
+ for(int index = 0; index < outputFeedNames.length; ++index) {
+ MapMessage m = (MapMessage) consumer.receive();
+ System.out.println("Consumed: " + m.toString());
+ assertMessage(m);
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_NAMES.getName()),
+ outputFeedNames[index]);
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
+ outputFeedPaths[index]);
}
- System.out.println("Consumed: " + m.toString());
- assertMessage(m);
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_NAMES.getName()), "raw-logs");
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
- "/raw-logs/10/05/05/00/20");
connection.close();
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/152afb1c/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java
----------------------------------------------------------------------
diff --git a/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java b/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java
index 2787d7f..f08949d 100644
--- a/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java
+++ b/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java
@@ -21,12 +21,14 @@ import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.falcon.workflow.FalconPostProcessing;
import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.commons.lang.StringUtils;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import javax.jms.*;
+import java.util.concurrent.CountDownLatch;
/**
* Test for validating the falcon post processing utility.
@@ -36,19 +38,20 @@ public class FalconPostProcessingTest {
private String[] args;
private static final String BROKER_URL = "vm://localhost?broker.useJmx=false&broker.persistent=true";
private static final String BROKER_IMPL_CLASS = "org.apache.activemq.ActiveMQConnectionFactory";
- private static final String FALCON_TOPIC_NAME = "FALCON.ENTITY.TOPIC";
private static final String ENTITY_NAME = "agg-coord";
private BrokerService broker;
private volatile AssertionError error;
+ private CountDownLatch latch = new CountDownLatch(1);
+ private String[] outputFeedNames = {"out-click-logs", "out-raw-logs"};
+ private String[] outputFeedPaths = {"/out-click-logs/10/05/05/00/20", "/out-raw-logs/10/05/05/00/20"};
@BeforeClass
public void setup() throws Exception {
args = new String[]{
"-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), ENTITY_NAME,
- "-" + WorkflowExecutionArgs.OUTPUT_FEED_NAMES.getName(), "out-click-logs,out-raw-logs",
- "-" + WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName(),
- "/out-click-logs/10/05/05/00/20,/out-raw-logs/10/05/05/00/20",
+ "-" + WorkflowExecutionArgs.OUTPUT_FEED_NAMES.getName(), StringUtils.join(outputFeedNames, ","),
+ "-" + WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName(), StringUtils.join(outputFeedPaths, ","),
"-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
"-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), "falcon",
"-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
@@ -95,8 +98,8 @@ public class FalconPostProcessingTest {
@Override
public void run() {
try {
- consumer(BROKER_URL, "FALCON." + ENTITY_NAME); // user message
- consumer(BROKER_URL, FALCON_TOPIC_NAME); // falcon message
+ // falcon message [FALCON_TOPIC_NAME] and user message ["FALCON." + ENTITY_NAME]
+ consumer(BROKER_URL, "FALCON.>");
} catch (AssertionError e) {
error = e;
} catch (JMSException ignore) {
@@ -105,7 +108,8 @@ public class FalconPostProcessingTest {
}
};
t.start();
- Thread.sleep(1500);
+
+ latch.await();
new FalconPostProcessing().run(this.args);
t.join();
if (error != null) {
@@ -122,20 +126,32 @@ public class FalconPostProcessingTest {
Destination destination = session.createTopic(topic);
MessageConsumer consumer = session.createConsumer(destination);
- // wait till you get atleast one message
- MapMessage m;
- for (m = null; m == null;) {
- m = (MapMessage) consumer.receive();
- }
- System.out.println("Consumed: " + m.toString());
+ latch.countDown();
+
+ // Verify user message
+ verifyMesssage(consumer);
+
+ // Verify falcon message
+ verifyMesssage(consumer);
- assertMessage(m);
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_NAMES.getName()), "out-click-logs");
- Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
- "/out-click-logs/10/05/05/00/20");
connection.close();
}
+ private void verifyMesssage(MessageConsumer consumer) throws JMSException {
+ for (int index = 0; index < outputFeedPaths.length; ++index) {
+ // receive call is blocking
+ MapMessage m = (MapMessage) consumer.receive();
+
+ System.out.println("Received JMS message {}" + m.toString());
+ System.out.println("Consumed: " + m.toString());
+ assertMessage(m);
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_NAMES.getName()),
+ outputFeedNames[index]);
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.OUTPUT_FEED_PATHS.getName()),
+ outputFeedPaths[index]);
+ }
+ }
+
private void assertMessage(MapMessage m) throws JMSException {
Assert.assertEquals(m.getString(WorkflowExecutionArgs.ENTITY_NAME.getName()), "agg-coord");
String workflowUser = m.getString(WorkflowExecutionArgs.WORKFLOW_USER.getName());
[3/5] git commit: FALCON-20 Remove dependency on custom InMobi
DistCp. Contributed by Sowmya Ramesh
Posted by ve...@apache.org.
FALCON-20 Remove dependency on custom InMobi DistCp. Contributed by Sowmya Ramesh
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/d82c01da
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/d82c01da
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/d82c01da
Branch: refs/heads/master
Commit: d82c01dafc51bba758f5c7be21e5b43219e12f2c
Parents: b0929b4
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Wed Oct 8 15:22:00 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Wed Oct 8 16:21:37 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 3 +++
common/src/main/resources/startup.properties | 2 +-
hadoop-dependencies/pom.xml | 6 +++++
hadoop-webapp/pom.xml | 6 +++++
.../action/feed/replication-action.xml | 5 +++-
pom.xml | 26 ++++++--------------
replication/pom.xml | 9 +++----
.../falcon/replication/CustomReplicator.java | 2 ++
.../falcon/replication/FeedReplicator.java | 12 ++++-----
.../replication/FilteredCopyListingTest.java | 11 +++++++--
src/conf/startup.properties | 2 +-
11 files changed, 48 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index dca9fa9..c5ba06c 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -35,6 +35,9 @@ Trunk (Unreleased)
FALCON-263 API to get workflow parameters. (pavan kumar kolamuri via Shwetha GS)
IMPROVEMENTS
+ FALCON-20 Remove dependency on custom InMobi DistCp (Sowmya Ramesh via
+ Venkatesh Seetharam)
+
FALCON-758 Discontinue support for Oozie-3.x (Peeyush Bishnoi via
Venkatesh Seetharam)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/common/src/main/resources/startup.properties
----------------------------------------------------------------------
diff --git a/common/src/main/resources/startup.properties b/common/src/main/resources/startup.properties
index 99aa3c0..433c2a8 100644
--- a/common/src/main/resources/startup.properties
+++ b/common/src/main/resources/startup.properties
@@ -49,7 +49,7 @@
*.broker.impl.class=org.apache.activemq.ActiveMQConnectionFactory
##### List of shared libraries for Falcon workflows #####
-*.shared.libs=activemq-core,geronimo-j2ee-management,hadoop-distcp,jms,json-simple,oozie-client,spring-jms
+*.shared.libs=activemq-core,geronimo-j2ee-management,jms,json-simple,oozie-client,spring-jms
##### Workflow Job Execution Completion listeners #####
*.workflow.execution.listeners=
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/hadoop-dependencies/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-dependencies/pom.xml b/hadoop-dependencies/pom.xml
index b3e3ebc..7b0aa1d 100644
--- a/hadoop-dependencies/pom.xml
+++ b/hadoop-dependencies/pom.xml
@@ -102,6 +102,12 @@
<artifactId>jersey-core</artifactId>
<scope>compile</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-distcp</artifactId>
+ <scope>compile</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/hadoop-webapp/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-webapp/pom.xml b/hadoop-webapp/pom.xml
index 1228074..236c269 100644
--- a/hadoop-webapp/pom.xml
+++ b/hadoop-webapp/pom.xml
@@ -87,6 +87,12 @@
</dependency>
<dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-distcp</artifactId>
+ <scope>compile</scope>
+ </dependency>
+
+ <dependency>
<groupId>com.sun.jersey</groupId>
<artifactId>jersey-client</artifactId>
<scope>compile</scope>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/oozie/src/main/resources/action/feed/replication-action.xml
----------------------------------------------------------------------
diff --git a/oozie/src/main/resources/action/feed/replication-action.xml b/oozie/src/main/resources/action/feed/replication-action.xml
index dffdd92..5d154b4 100644
--- a/oozie/src/main/resources/action/feed/replication-action.xml
+++ b/oozie/src/main/resources/action/feed/replication-action.xml
@@ -33,6 +33,10 @@
<name>oozie.launcher.mapred.job.priority</name>
<value>${jobPriority}</value>
</property>
+ <property> <!-- DistCp jars -->
+ <name>oozie.action.sharelib.for.java</name>
+ <value>distcp</value>
+ </property>
</configuration>
<main-class>org.apache.falcon.replication.FeedReplicator</main-class>
<arg>-Dfalcon.include.path=${sourceRelativePaths}</arg>
@@ -48,7 +52,6 @@
<arg>${distcpTargetPaths}</arg>
<arg>-falconFeedStorageType</arg>
<arg>${falconFeedStorageType}</arg>
- <file>${wf:conf("falcon.libpath")}/hadoop-distcp.jar</file>
</java>
<ok to="succeeded-post-processing"/>
<error to="failed-post-processing"/>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 02473b1..ed046a6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -109,7 +109,6 @@
<activemq.version>5.4.3</activemq.version>
<hive.version>0.11.0</hive.version>
<hcatalog.version>0.11.0</hcatalog.version>
- <hadoop-distcp.version>0.11</hadoop-distcp.version>
<jetty.version>6.1.26</jetty.version>
<jersey.version>1.9</jersey.version>
<internal.maven.repo>file:///tmp/falcontemprepo</internal.maven.repo>
@@ -236,6 +235,13 @@
</exclusion>
</exclusions>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-distcp</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>provided</scope>
+ </dependency>
</dependencies>
</dependencyManagement>
</profile>
@@ -779,24 +785,6 @@
</dependency>
<dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-auth</artifactId>
- <version>${hadoop.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.hadoop.inmobi.tools</groupId>
- <artifactId>hadoop-distcp</artifactId>
- <version>${hadoop-distcp.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-core</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
<groupId>org.apache.falcon</groupId>
<artifactId>falcon-messaging</artifactId>
<version>${project.version}</version>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/replication/pom.xml
----------------------------------------------------------------------
diff --git a/replication/pom.xml b/replication/pom.xml
index 9dba461..47e5f6a 100644
--- a/replication/pom.xml
+++ b/replication/pom.xml
@@ -42,6 +42,10 @@
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-distcp</artifactId>
+ </dependency>
</dependencies>
</profile>
</profiles>
@@ -58,11 +62,6 @@
</dependency>
<dependency>
- <groupId>org.apache.hadoop.inmobi.tools</groupId>
- <artifactId>hadoop-distcp</artifactId>
- </dependency>
-
- <dependency>
<groupId>org.testng</groupId>
<artifactId>testng</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/replication/src/main/java/org/apache/falcon/replication/CustomReplicator.java
----------------------------------------------------------------------
diff --git a/replication/src/main/java/org/apache/falcon/replication/CustomReplicator.java b/replication/src/main/java/org/apache/falcon/replication/CustomReplicator.java
index 164bfb0..d99fe20 100644
--- a/replication/src/main/java/org/apache/falcon/replication/CustomReplicator.java
+++ b/replication/src/main/java/org/apache/falcon/replication/CustomReplicator.java
@@ -36,6 +36,7 @@ import java.io.IOException;
public class CustomReplicator extends DistCp {
private static final Logger LOG = LoggerFactory.getLogger(CustomReplicator.class);
+ private final DistCpOptions inputOptions;
/**
* Public Constructor. Creates DistCp object with specified input-parameters.
@@ -47,6 +48,7 @@ public class CustomReplicator extends DistCp {
*/
public CustomReplicator(Configuration configuration, DistCpOptions inputOptions) throws Exception {
super(configuration, inputOptions);
+ this.inputOptions = inputOptions;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java
----------------------------------------------------------------------
diff --git a/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java b/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java
index b6d2fa3..d927fff 100644
--- a/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java
+++ b/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java
@@ -20,7 +20,6 @@ package org.apache.falcon.replication;
import org.apache.commons.cli.*;
import org.apache.commons.lang.StringUtils;
import org.apache.falcon.entity.EntityUtil;
-import org.apache.falcon.entity.Storage;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileStatus;
@@ -44,6 +43,7 @@ import java.util.regex.Pattern;
public class FeedReplicator extends Configured implements Tool {
private static final Logger LOG = LoggerFactory.getLogger(FeedReplicator.class);
+ private static final String IGNORE = "IGNORE";
public static void main(String[] args) throws Exception {
ToolRunner.run(new Configuration(), new FeedReplicator(), args);
@@ -62,16 +62,15 @@ public class FeedReplicator extends Configured implements Tool {
LOG.info("{} found conf ? {}", confPath, confPath.getFileSystem(conf).exists(confPath));
conf.addResource(confPath);
- String falconFeedStorageType = cmd.getOptionValue("falconFeedStorageType").trim();
- Storage.TYPE feedStorageType = Storage.TYPE.valueOf(falconFeedStorageType);
+ final boolean includePathSet = !IGNORE.equalsIgnoreCase(conf.get("falcon.include.path"));
- DistCp distCp = (feedStorageType == Storage.TYPE.FILESYSTEM)
+ DistCp distCp = (includePathSet)
? new CustomReplicator(conf, options)
: new DistCp(conf, options);
LOG.info("Started DistCp");
distCp.execute();
- if (feedStorageType == Storage.TYPE.FILESYSTEM) {
+ if (includePathSet) {
executePostProcessing(options); // this only applies for FileSystem Storage.
}
@@ -116,8 +115,7 @@ public class FeedReplicator extends Configured implements Tool {
distcpOptions.setSyncFolder(true);
distcpOptions.setBlocking(true);
distcpOptions.setMaxMaps(Integer.valueOf(cmd.getOptionValue("maxMaps")));
- distcpOptions.setMapBandwidthKB(Integer.valueOf(cmd.getOptionValue("mapBandwidthKB")));
-
+ distcpOptions.setMapBandwidth(Integer.valueOf(cmd.getOptionValue("mapBandwidthKB")));
return distcpOptions;
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/replication/src/test/java/org/apache/falcon/replication/FilteredCopyListingTest.java
----------------------------------------------------------------------
diff --git a/replication/src/test/java/org/apache/falcon/replication/FilteredCopyListingTest.java b/replication/src/test/java/org/apache/falcon/replication/FilteredCopyListingTest.java
index e308866..09e343b 100644
--- a/replication/src/test/java/org/apache/falcon/replication/FilteredCopyListingTest.java
+++ b/replication/src/test/java/org/apache/falcon/replication/FilteredCopyListingTest.java
@@ -26,6 +26,7 @@ import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.tools.CopyListingFileStatus;
import org.apache.hadoop.tools.DistCpOptions;
import org.apache.hadoop.tools.util.DistCpUtils;
import org.testng.Assert;
@@ -117,6 +118,7 @@ public class FilteredCopyListingTest {
Path target = new Path(fileSystemPath.toString() + "///tmp/target");
Path listingPath = new Path(fileSystemPath.toString() + "///tmp/META/fileList.seq");
DistCpOptions options = new DistCpOptions(Arrays.asList(source), target);
+ options.setSyncFolder(true);
new FilteredCopyListing(new Configuration(), CREDENTIALS).buildListing(listingPath, options);
@@ -132,6 +134,7 @@ public class FilteredCopyListingTest {
Path target = new Path(fileSystemPath.toString() + "///tmp/target");
Path listingPath = new Path(fileSystemPath.toString() + "///tmp/META/fileList.seq");
DistCpOptions options = new DistCpOptions(Arrays.asList(source), target);
+ options.setSyncFolder(true);
Configuration configuration = new Configuration();
configuration.set("falcon.include.path", "*/3/*");
@@ -149,6 +152,7 @@ public class FilteredCopyListingTest {
Path target = new Path(fileSystemPath.toString() + "///tmp/target");
Path listingPath = new Path(fileSystemPath.toString() + "///tmp/META/fileList.seq");
DistCpOptions options = new DistCpOptions(Arrays.asList(source), target);
+ options.setSyncFolder(true);
Configuration configuration = new Configuration();
configuration.set("falcon.include.path", "*/3/?");
@@ -166,6 +170,7 @@ public class FilteredCopyListingTest {
Path target = new Path(fileSystemPath.toString() + "///tmp/target");
Path listingPath = new Path(fileSystemPath.toString() + "///tmp/META/fileList.seq");
DistCpOptions options = new DistCpOptions(Arrays.asList(source), target);
+ options.setSyncFolder(true);
Configuration configuration = new Configuration();
configuration.set("falcon.include.path", "*/3/[47]");
@@ -183,6 +188,7 @@ public class FilteredCopyListingTest {
Path target = new Path(fileSystemPath.toString() + "///tmp/target");
Path listingPath = new Path(fileSystemPath.toString() + "///tmp/META/fileList.seq");
DistCpOptions options = new DistCpOptions(Arrays.asList(source), target);
+ options.setSyncFolder(true);
Configuration configuration = new Configuration();
configuration.set("falcon.include.path", "*/3/40");
@@ -200,6 +206,7 @@ public class FilteredCopyListingTest {
Path target = new Path(fileSystemPath.toString() + "///tmp/target");
Path listingPath = new Path(fileSystemPath.toString() + "///tmp/META/fileList.seq");
DistCpOptions options = new DistCpOptions(Arrays.asList(source), target);
+ options.setSyncFolder(true);
Configuration configuration = new Configuration();
configuration.set("falcon.include.path", "*/3/{4,7}");
@@ -212,13 +219,13 @@ public class FilteredCopyListingTest {
SequenceFile.Reader reader = new SequenceFile.Reader(FileSystem.getLocal(new Configuration()),
listingPath, new Configuration());
Text key = new Text();
- FileStatus value = new FileStatus();
+ FileStatus value = new CopyListingFileStatus();
Map<String, String> actualValues = new HashMap<String, String>();
while (reader.next(key, value)) {
actualValues.put(value.getPath().toString(), key.toString());
}
- Assert.assertEquals(expected == -1 ? EXPECTED_VALUES.size() : expected, actualValues.size());
+ Assert.assertEquals(actualValues.size(), expected == -1 ? EXPECTED_VALUES.size() : expected);
for (Map.Entry<String, String> entry : actualValues.entrySet()) {
Assert.assertEquals(entry.getValue(), EXPECTED_VALUES.get(entry.getKey()));
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d82c01da/src/conf/startup.properties
----------------------------------------------------------------------
diff --git a/src/conf/startup.properties b/src/conf/startup.properties
index 78466af..2db4b1e 100644
--- a/src/conf/startup.properties
+++ b/src/conf/startup.properties
@@ -61,7 +61,7 @@ prism.configstore.listeners=org.apache.falcon.entity.v0.EntityGraph,\
*.broker.impl.class=org.apache.activemq.ActiveMQConnectionFactory
##### List of shared libraries for Falcon workflows #####
-*.shared.libs=activemq-core,ant,geronimo-j2ee-management,hadoop-distcp,jms,json-simple,oozie-client,spring-jms
+*.shared.libs=activemq-core,ant,geronimo-j2ee-management,jms,json-simple,oozie-client,spring-jms
##### Workflow Job Execution Completion listeners #####
*.workflow.execution.listeners=
[5/5] git commit: FALCON-786 FalconAuditFilter - Arguments in wrong
order. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-786 FalconAuditFilter - Arguments in wrong order. Contributed by Venkatesh Seetharam
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/396db782
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/396db782
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/396db782
Branch: refs/heads/master
Commit: 396db78255f03d9f7d5a04190160445da1fb96d2
Parents: 43110bd
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Wed Oct 8 16:44:20 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Wed Oct 8 16:44:20 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 2 ++
.../main/java/org/apache/falcon/security/FalconAuditFilter.java | 2 +-
2 files changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/396db782/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index d7571ba..9f79151 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -115,6 +115,8 @@ Trunk (Unreleased)
OPTIMIZATIONS
BUG FIXES
+ FALCON-786 FalconAuditFilter - Arguments in wrong order (Venkatesh Seetharam)
+
FALCON-789 Post processing is broken for Feeds (Sowmya Ramesh via
Venkatesh Seetharam)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/396db782/prism/src/main/java/org/apache/falcon/security/FalconAuditFilter.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/security/FalconAuditFilter.java b/prism/src/main/java/org/apache/falcon/security/FalconAuditFilter.java
index d8b73b2..21e28b3 100644
--- a/prism/src/main/java/org/apache/falcon/security/FalconAuditFilter.java
+++ b/prism/src/main/java/org/apache/falcon/security/FalconAuditFilter.java
@@ -68,7 +68,7 @@ public class FalconAuditFilter implements Filter {
LOG.debug("Audit: {}/{} performed request {} ({}) at time {}",
who, fromAddress, whatURL, whatAddrs, whenISO9601);
- GenericAlert.audit(who, fromHost, fromAddress, whatURL, whatAddrs, whenISO9601);
+ GenericAlert.audit(who, fromAddress, fromHost, whatURL, whatAddrs, whenISO9601);
}
private String getUserFromRequest(HttpServletRequest httpRequest) {