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/08/08 22:31:34 UTC
[1/8] git commit: FALCON-484 - Refactor workflow related code into
context, listener and Notification. Contributed by Venkatesh Seetharam
Repository: incubator-falcon
Updated Branches:
refs/heads/master 522788157 -> 9921457c3
FALCON-484 - Refactor workflow related code into context, listener and Notification. 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/b42c53ba
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/b42c53ba
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/b42c53ba
Branch: refs/heads/master
Commit: b42c53baaf29fee4466264aac83b70dca0903af9
Parents: 5227881
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Fri Aug 8 12:55:48 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Fri Aug 8 12:55:48 2014 -0700
----------------------------------------------------------------------
.../falcon/workflow/WorkflowExecutionArgs.java | 122 +++++++
.../workflow/WorkflowExecutionContext.java | 326 +++++++++++++++++++
.../workflow/WorkflowExecutionListener.java | 31 ++
.../WorkflowJobEndNotificationService.java | 112 +++++++
common/src/main/resources/startup.properties | 1 +
.../workflow/WorkflowExecutionContextTest.java | 301 +++++++++++++++++
src/conf/startup.properties | 1 +
7 files changed, 894 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b42c53ba/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionArgs.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionArgs.java b/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionArgs.java
new file mode 100644
index 0000000..92af3e1
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionArgs.java
@@ -0,0 +1,122 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.workflow;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+
+/**
+ * Arguments for workflow execution.
+ */
+public enum WorkflowExecutionArgs {
+
+ // instance details
+ NOMINAL_TIME("nominalTime", "instance time"),
+ ENTITY_TYPE("entityType", "type of the entity"),
+ ENTITY_NAME("entityName", "name of the entity"),
+ TIMESTAMP("timeStamp", "current timestamp"),
+
+ // where
+ CLUSTER_NAME("cluster", "name of the current cluster"),
+ OPERATION("operation", "operation like generate, delete, replicate"),
+
+ // who
+ WORKFLOW_USER("workflowUser", "user who owns the feed instance (partition)"),
+
+ // what
+ // workflow details
+ USER_WORKFLOW_ENGINE("userWorkflowEngine", "user workflow engine type", false),
+ USER_WORKFLOW_NAME("userWorkflowName", "user workflow name", false),
+ USER_WORKFLOW_VERSION("userWorkflowVersion", "user workflow version", false),
+
+ // workflow execution details
+ WORKFLOW_ID("workflowId", "current workflow-id of the instance"),
+ RUN_ID("runId", "current run-id of the instance"),
+ STATUS("status", "status of the user workflow isnstance"),
+ WF_ENGINE_URL("workflowEngineUrl", "url of workflow engine server, ex:oozie", false),
+ USER_SUBFLOW_ID("subflowId", "external id of user workflow", false),
+
+ WF_START_TIME("workflowStartTime", "workflow start time", false),
+ WF_END_TIME("workflowEndTime", "workflow end time", false),
+ WF_DURATION("workflowDuration", "workflow duration", false),
+
+ // what inputs
+ INPUT_FEED_NAMES("falconInputFeeds", "name of the feeds which are used as inputs", false),
+ INPUT_FEED_PATHS("falconInputPaths", "comma separated input feed instance paths", false),
+
+ // what outputs
+ FEED_NAMES("feedNames", "name of the feeds which are generated/replicated/deleted"),
+ FEED_INSTANCE_PATHS("feedInstancePaths", "comma separated feed instance paths"),
+
+ // broker related parameters
+ TOPIC_NAME("topicName", "name of the topic to be used to send JMS message", false),
+ BRKR_IMPL_CLASS("brokerImplClass", "falcon message broker Implementation class"),
+ BRKR_URL("brokerUrl", "falcon message broker url"),
+ USER_BRKR_IMPL_CLASS("userBrokerImplClass", "user broker Impl class", false),
+ USER_BRKR_URL("userBrokerUrl", "user broker url", false),
+ BRKR_TTL("brokerTTL", "time to live for broker message in sec", false),
+
+ // state maintained
+ LOG_FILE("logFile", "log file path where feeds to be deleted are recorded"),
+ // execution context data recorded
+ LOG_DIR("logDir", "log dir where lineage can be recorded"),
+
+ CONTEXT_FILE("contextFile", "wf execution context file path where wf properties are recorded", false),
+ CONTEXT_TYPE("contextType", "wf execution context type, pre or post processing", false);
+
+
+ private final String name;
+ private final String description;
+ private final boolean isRequired;
+
+ WorkflowExecutionArgs(String name, String description) {
+ this(name, description, true);
+ }
+
+ WorkflowExecutionArgs(String name, String description, boolean isRequired) {
+ this.name = name;
+ this.description = description;
+ this.isRequired = isRequired;
+ }
+
+ public Option getOption() {
+ return new Option(this.name, true, this.description);
+ }
+
+ public String getName() {
+ return this.name;
+ }
+
+ public String getDescription() {
+ return description;
+ }
+
+ public boolean isRequired() {
+ return isRequired;
+ }
+
+ public String getOptionValue(CommandLine cmd) {
+ return cmd.getOptionValue(this.name);
+ }
+
+ @Override
+ public String toString() {
+ return getName();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b42c53ba/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
new file mode 100644
index 0000000..60e6c58
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionContext.java
@@ -0,0 +1,326 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.workflow;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.GnuParser;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.lang.StringUtils;
+import org.apache.falcon.FalconException;
+import org.apache.falcon.entity.v0.SchemaHelper;
+import org.apache.falcon.hadoop.HadoopClientFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Captures the workflow execution context.
+ */
+public class WorkflowExecutionContext {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WorkflowExecutionContext.class);
+
+ public static final String PROCESS_INSTANCE_FORMAT = "yyyy-MM-dd-HH-mm"; // nominal time
+
+ public static final String OUTPUT_FEED_SEPARATOR = ",";
+ public static final String INPUT_FEED_SEPARATOR = "#";
+
+
+ /**
+ * Workflow execution status.
+ */
+ public enum Status {SUCCEEDED, FAILED}
+
+ /**
+ * Workflow execution type.
+ */
+ public enum Type {PRE_PROCESSING, POST_PROCESSING}
+
+ /**
+ * Entity operations supported.
+ */
+ public enum EntityOperations {
+ GENERATE, DELETE, ARCHIVE, REPLICATE, CHMOD
+ }
+
+ private final Map<WorkflowExecutionArgs, String> context;
+ private final long creationTime;
+
+ protected WorkflowExecutionContext(Map<WorkflowExecutionArgs, String> context) {
+ this.context = context;
+ creationTime = System.currentTimeMillis();
+ }
+
+ public String getValue(WorkflowExecutionArgs arg) {
+ return context.get(arg);
+ }
+
+ public String getValue(WorkflowExecutionArgs arg, String defaultValue) {
+ return context.containsKey(arg) ? context.get(arg) : defaultValue;
+ }
+
+ public boolean containsKey(WorkflowExecutionArgs arg) {
+ return context.containsKey(arg);
+ }
+
+ public Set<Map.Entry<WorkflowExecutionArgs, String>> entrySet() {
+ return context.entrySet();
+ }
+
+ // helper methods
+ public boolean hasWorkflowSucceeded() {
+ return Status.SUCCEEDED.name().equals(getValue(WorkflowExecutionArgs.STATUS));
+ }
+
+ public boolean hasWorkflowFailed() {
+ return Status.FAILED.name().equals(getValue(WorkflowExecutionArgs.STATUS));
+ }
+
+ public String getContextFile() {
+ return getValue(WorkflowExecutionArgs.CONTEXT_FILE);
+ }
+
+ public String getLogDir() {
+ return getValue(WorkflowExecutionArgs.LOG_DIR);
+ }
+
+ public String getLogFile() {
+ return getValue(WorkflowExecutionArgs.LOG_FILE);
+ }
+
+ String getNominalTime() {
+ return getValue(WorkflowExecutionArgs.NOMINAL_TIME);
+ }
+
+ /**
+ * Returns nominal time as a ISO8601 formatted string.
+ * @return a ISO8601 formatted string
+ */
+ public String getNominalTimeAsISO8601() {
+ return SchemaHelper.formatDateUTCToISO8601(getNominalTime(), PROCESS_INSTANCE_FORMAT);
+ }
+
+ public String getTimestamp() {
+ return getValue(WorkflowExecutionArgs.TIMESTAMP);
+ }
+
+ /**
+ * Returns timestamp as a ISO8601 formatted string.
+ * @return a ISO8601 formatted string
+ */
+ public String getTimeStampAsISO8601() {
+ return SchemaHelper.formatDateUTCToISO8601(getTimestamp(), PROCESS_INSTANCE_FORMAT);
+ }
+
+ public String getClusterName() {
+ return getValue(WorkflowExecutionArgs.CLUSTER_NAME);
+ }
+
+ public String getEntityName() {
+ return getValue(WorkflowExecutionArgs.ENTITY_NAME);
+ }
+
+ public String getEntityType() {
+ return getValue(WorkflowExecutionArgs.ENTITY_TYPE);
+ }
+
+ public EntityOperations getOperation() {
+ return EntityOperations.valueOf(getValue(WorkflowExecutionArgs.OPERATION));
+ }
+
+ public String getOutputFeedNames() {
+ return getValue(WorkflowExecutionArgs.FEED_NAMES);
+ }
+
+ public String[] getOutputFeedNamesList() {
+ return getOutputFeedNames().split(OUTPUT_FEED_SEPARATOR);
+ }
+
+ public String getOutputFeedInstancePaths() {
+ return getValue(WorkflowExecutionArgs.FEED_INSTANCE_PATHS);
+ }
+
+ public String[] getOutputFeedInstancePathsList() {
+ return getOutputFeedInstancePaths().split(OUTPUT_FEED_SEPARATOR);
+ }
+
+ public String getInputFeedNames() {
+ return getValue(WorkflowExecutionArgs.INPUT_FEED_NAMES);
+ }
+
+ public String[] getInputFeedNamesList() {
+ return getInputFeedNames().split(INPUT_FEED_SEPARATOR);
+ }
+
+ public String getInputFeedInstancePaths() {
+ return getValue(WorkflowExecutionArgs.INPUT_FEED_PATHS);
+ }
+
+ public String[] getInputFeedInstancePathsList() {
+ return getInputFeedInstancePaths().split(INPUT_FEED_SEPARATOR);
+ }
+
+ public String getWorkflowEngineUrl() {
+ return getValue(WorkflowExecutionArgs.WF_ENGINE_URL);
+ }
+
+ public String getUserWorkflowEngine() {
+ return getValue(WorkflowExecutionArgs.USER_WORKFLOW_ENGINE);
+ }
+
+ public String getUserWorkflowVersion() {
+ return getValue(WorkflowExecutionArgs.USER_WORKFLOW_VERSION);
+ }
+
+ public String getWorkflowId() {
+ return getValue(WorkflowExecutionArgs.WORKFLOW_ID);
+ }
+
+ public String getUserSubflowId() {
+ return getValue(WorkflowExecutionArgs.USER_SUBFLOW_ID);
+ }
+
+ public int getWorkflowRunId() {
+ return Integer.parseInt(getValue(WorkflowExecutionArgs.RUN_ID));
+ }
+
+ public String getWorkflowRunIdString() {
+ return String.valueOf(Integer.parseInt(getValue(WorkflowExecutionArgs.RUN_ID)));
+ }
+
+ public String getWorkflowUser() {
+ return getValue(WorkflowExecutionArgs.WORKFLOW_USER);
+ }
+
+ public long getExecutionCompletionTime() {
+ return creationTime;
+ }
+
+ /**
+ * this method is invoked from with in the workflow.
+ *
+ * @throws java.io.IOException
+ * @throws org.apache.falcon.FalconException
+ */
+ public void serialize() throws IOException, FalconException {
+ serialize(getContextFile());
+ }
+
+ /**
+ * this method is invoked from with in the workflow.
+ *
+ * @param contextFile file to serialize the workflow execution metadata
+ * @throws org.apache.falcon.FalconException
+ */
+ public void serialize(String contextFile) throws FalconException {
+ LOG.info("Saving context to: [{}]", contextFile);
+ OutputStream out = null;
+ Path file = new Path(contextFile);
+ try {
+ FileSystem fs = HadoopClientFactory.get().createFileSystem(file.toUri());
+ out = fs.create(file);
+
+ // making sure falcon can read this file
+ FsPermission permission = new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL);
+ fs.setPermission(file, permission);
+
+ out.write(JSONValue.toJSONString(context).getBytes());
+ } catch (IOException e) {
+ throw new FalconException("Error serializing context to: " + contextFile, e);
+ } finally {
+ if (out != null) {
+ try {
+ out.close();
+ } catch (IOException ignore) {
+ // ignore
+ }
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public static WorkflowExecutionContext deSerialize(String contextFile) throws FalconException {
+ try {
+ Path lineageDataPath = new Path(contextFile); // file has 777 permissions
+ FileSystem fs = HadoopClientFactory.get().createFileSystem(lineageDataPath.toUri());
+
+ BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(lineageDataPath)));
+ return new WorkflowExecutionContext((Map<WorkflowExecutionArgs, String>) JSONValue.parse(in));
+ } catch (IOException e) {
+ throw new FalconException("Error opening context file: " + contextFile, e);
+ }
+ }
+
+ public static String getFilePath(String logDir, String entityName) {
+ // LOG_DIR is sufficiently unique
+ return new Path(logDir, entityName + "-wf-post-exec-context.json").toString();
+ }
+
+ public static WorkflowExecutionContext create(String[] args, Type type) throws FalconException {
+ try {
+ Map<WorkflowExecutionArgs, String> wfProperties = new HashMap<WorkflowExecutionArgs, String>();
+ wfProperties.put(WorkflowExecutionArgs.CONTEXT_TYPE, type.name());
+
+ CommandLine cmd = getCommand(args);
+ for (WorkflowExecutionArgs arg : WorkflowExecutionArgs.values()) {
+ String optionValue = arg.getOptionValue(cmd);
+ if (StringUtils.isNotEmpty(optionValue)) {
+ wfProperties.put(arg, optionValue);
+ }
+ }
+
+ wfProperties.put(WorkflowExecutionArgs.CONTEXT_FILE,
+ getFilePath(wfProperties.get(WorkflowExecutionArgs.LOG_DIR),
+ wfProperties.get(WorkflowExecutionArgs.ENTITY_NAME)));
+ return new WorkflowExecutionContext(wfProperties);
+ } catch (ParseException e) {
+ throw new FalconException("Error parsing wf args", e);
+ }
+ }
+
+ private static CommandLine getCommand(String[] arguments) throws ParseException {
+ Options options = new Options();
+
+ for (WorkflowExecutionArgs arg : WorkflowExecutionArgs.values()) {
+ addOption(options, arg, arg.isRequired());
+ }
+
+ return new GnuParser().parse(options, arguments, false);
+ }
+
+ private static void addOption(Options options, WorkflowExecutionArgs arg, boolean isRequired) {
+ Option option = arg.getOption();
+ option.setRequired(isRequired);
+ options.addOption(option);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b42c53ba/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionListener.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionListener.java b/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionListener.java
new file mode 100644
index 0000000..2d3a477
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionListener.java
@@ -0,0 +1,31 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.workflow;
+
+import org.apache.falcon.FalconException;
+
+/**
+ * A listener interface for workflow execution.
+ */
+public interface WorkflowExecutionListener {
+
+ void onSuccess(WorkflowExecutionContext context) throws FalconException;
+
+ void onFailure(WorkflowExecutionContext context) throws FalconException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b42c53ba/common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java b/common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java
new file mode 100644
index 0000000..0602214
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java
@@ -0,0 +1,112 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.workflow;
+
+import org.apache.falcon.FalconException;
+import org.apache.falcon.aspect.GenericAlert;
+import org.apache.falcon.entity.v0.SchemaHelper;
+import org.apache.falcon.resource.InstancesResult;
+import org.apache.falcon.security.CurrentUser;
+import org.apache.falcon.service.FalconService;
+import org.apache.falcon.workflow.engine.AbstractWorkflowEngine;
+
+import java.util.Date;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+/**
+ * A workflow job end notification service.
+ */
+public class WorkflowJobEndNotificationService implements FalconService {
+
+ public static final String NAME = WorkflowJobEndNotificationService.class.getSimpleName();
+
+ private Set<WorkflowExecutionListener> listeners = new LinkedHashSet<WorkflowExecutionListener>();
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public void init() throws FalconException {
+ // do nothing
+ }
+
+ @Override
+ public void destroy() throws FalconException {
+ listeners.clear();
+ }
+
+ public void registerListener(WorkflowExecutionListener listener) {
+ listeners.add(listener);
+ }
+
+ public void unregisterListener(WorkflowExecutionListener listener) {
+ listeners.remove(listener);
+ }
+
+ public void notifyFailure(WorkflowExecutionContext context) throws FalconException {
+ for (WorkflowExecutionListener listener : listeners) {
+ listener.onFailure(context);
+ }
+
+ instrumentAlert(context);
+ }
+
+ public void notifySuccess(WorkflowExecutionContext context) throws FalconException {
+ for (WorkflowExecutionListener listener : listeners) {
+ listener.onSuccess(context);
+ }
+
+ instrumentAlert(context);
+ }
+
+ private void instrumentAlert(WorkflowExecutionContext context) throws FalconException {
+ String clusterName = context.getClusterName();
+ String entityName = context.getEntityName();
+ String entityType = context.getEntityType();
+ String operation = context.getOperation().name();
+ String workflowId = context.getWorkflowId();
+ String workflowUser = context.getWorkflowUser();
+ String nominalTime = context.getNominalTimeAsISO8601();
+ String runId = String.valueOf(context.getWorkflowRunId());
+
+ CurrentUser.authenticate(context.getWorkflowUser());
+ AbstractWorkflowEngine wfEngine = WorkflowEngineFactory.getWorkflowEngine();
+ InstancesResult result = wfEngine.getJobDetails(clusterName, workflowId);
+ Date startTime = result.getInstances()[0].startTime;
+ Date endTime = result.getInstances()[0].endTime;
+ Long duration = (endTime.getTime() - startTime.getTime()) * 1000000;
+
+ try {
+ if (context.hasWorkflowFailed()) {
+ GenericAlert.instrumentFailedInstance(clusterName, entityType,
+ entityName, nominalTime, workflowId, workflowUser, runId, operation,
+ SchemaHelper.formatDateUTC(startTime), "", "", duration);
+ } else {
+ GenericAlert.instrumentSucceededInstance(clusterName, entityType,
+ entityName, nominalTime, workflowId, workflowUser, runId, operation,
+ SchemaHelper.formatDateUTC(startTime), duration);
+ }
+ } catch (Exception e) {
+ throw new FalconException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b42c53ba/common/src/main/resources/startup.properties
----------------------------------------------------------------------
diff --git a/common/src/main/resources/startup.properties b/common/src/main/resources/startup.properties
index 0d49b4b..d215a62 100644
--- a/common/src/main/resources/startup.properties
+++ b/common/src/main/resources/startup.properties
@@ -29,6 +29,7 @@
*.catalog.service.impl=org.apache.falcon.catalog.HiveCatalogService
*.application.services=org.apache.falcon.security.AuthenticationInitializationService,\
+ org.apache.falcon.workflow.WorkflowJobEndNotificationService, \
org.apache.falcon.service.ProcessSubscriberService,\
org.apache.falcon.entity.store.ConfigurationStore,\
org.apache.falcon.rerun.service.RetryService,\
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b42c53ba/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
new file mode 100644
index 0000000..93d8831
--- /dev/null
+++ b/common/src/test/java/org/apache/falcon/workflow/WorkflowExecutionContextTest.java
@@ -0,0 +1,301 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.workflow;
+
+import org.apache.falcon.entity.v0.SchemaHelper;
+import org.apache.falcon.entity.v0.process.EngineType;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+
+/**
+ * A test for WorkflowExecutionContext.
+ */
+public class WorkflowExecutionContextTest {
+
+ private static final String FALCON_USER = "falcon-user";
+ private static final String LOGS_DIR = "target/log";
+ private static final String NOMINAL_TIME = "2014-01-01-01-00";
+ private static final String OPERATION = "GENERATE";
+
+ private static final String CLUSTER_NAME = "primary-cluster";
+ private static final String ENTITY_NAME = "sample-process";
+ private static final String WORKFLOW_NAME = "imp-click-join-workflow";
+ private static final String WORKFLOW_VERSION = "1.0.9";
+
+ private static final String INPUT_FEED_NAMES = "impression-feed#clicks-feed";
+ private static final String INPUT_INSTANCE_PATHS =
+ "jail://global:00/falcon/impression-feed/2014/01/01,jail://global:00/falcon/impression-feed/2014/01/02"
+ + "#jail://global:00/falcon/clicks-feed/2014-01-01";
+
+ private static final String OUTPUT_FEED_NAMES = "imp-click-join1,imp-click-join2";
+ private static final String OUTPUT_INSTANCE_PATHS =
+ "jail://global:00/falcon/imp-click-join1/20140101,jail://global:00/falcon/imp-click-join2/20140101";
+
+ private static final String BROKER = "org.apache.activemq.ActiveMQConnectionFactory";
+
+ private static final String ISO8601_TIME = SchemaHelper.formatDateUTCToISO8601(
+ NOMINAL_TIME, WorkflowExecutionContext.PROCESS_INSTANCE_FORMAT);
+
+ private WorkflowExecutionContext context;
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ context = WorkflowExecutionContext.create(getTestMessageArgs(),
+ WorkflowExecutionContext.Type.POST_PROCESSING);
+ }
+
+ @Test
+ public void testGetValue() throws Exception {
+ Assert.assertEquals(context.getValue(WorkflowExecutionArgs.ENTITY_NAME), ENTITY_NAME);
+ }
+
+ @Test
+ public void testGetValueWithDefault() throws Exception {
+ Assert.assertEquals(context.getValue(WorkflowExecutionArgs.TOPIC_NAME, "ABSENT"), "ABSENT");
+ }
+
+ @Test
+ public void testContainsKey() throws Exception {
+ Assert.assertTrue(context.containsKey(WorkflowExecutionArgs.ENTITY_NAME));
+ Assert.assertFalse(context.containsKey(WorkflowExecutionArgs.TOPIC_NAME));
+ }
+
+ @Test
+ public void testEntrySet() throws Exception {
+ Assert.assertNotNull(context.entrySet());
+ }
+
+ @Test
+ public void testHasWorkflowSucceeded() throws Exception {
+ Assert.assertTrue(context.hasWorkflowSucceeded());
+ }
+
+ @Test
+ public void testHasWorkflowFailed() throws Exception {
+ Assert.assertFalse(context.hasWorkflowFailed());
+ }
+
+ @Test
+ public void testGetContextFile() throws Exception {
+ Assert.assertEquals(context.getContextFile(),
+ WorkflowExecutionContext.getFilePath(context.getLogDir(), context.getEntityName()));
+ }
+
+ @Test
+ public void testGetLogDir() throws Exception {
+ Assert.assertEquals(context.getLogDir(), LOGS_DIR);
+ }
+
+ @Test
+ public void testGetLogFile() throws Exception {
+ Assert.assertEquals(context.getLogFile(), LOGS_DIR + "/log.txt");
+ }
+
+ @Test
+ public void testGetNominalTime() throws Exception {
+ Assert.assertEquals(context.getNominalTime(), NOMINAL_TIME);
+ }
+
+ @Test
+ public void testGetNominalTimeAsISO8601() throws Exception {
+ Assert.assertEquals(context.getNominalTimeAsISO8601(), ISO8601_TIME);
+ }
+
+ @Test
+ public void testGetTimestamp() throws Exception {
+ Assert.assertEquals(context.getTimestamp(), NOMINAL_TIME);
+ }
+
+ @Test
+ public void testGetTimeStampAsISO8601() throws Exception {
+ Assert.assertEquals(context.getTimeStampAsISO8601(), ISO8601_TIME);
+ }
+
+ @Test
+ public void testGetClusterName() throws Exception {
+ Assert.assertEquals(context.getClusterName(), CLUSTER_NAME);
+ }
+
+ @Test
+ public void testGetEntityName() throws Exception {
+ Assert.assertEquals(context.getEntityName(), ENTITY_NAME);
+ }
+
+ @Test
+ public void testGetEntityType() throws Exception {
+ Assert.assertEquals(context.getEntityType(), "process");
+ }
+
+ @Test
+ public void testGetOperation() throws Exception {
+ Assert.assertEquals(context.getOperation().name(), OPERATION);
+ }
+
+ @Test
+ public void testGetOutputFeedNames() throws Exception {
+ Assert.assertEquals(context.getOutputFeedNames(), OUTPUT_FEED_NAMES);
+ }
+
+ @Test
+ public void testGetOutputFeedNamesList() throws Exception {
+ Assert.assertEquals(context.getOutputFeedNamesList(),
+ OUTPUT_FEED_NAMES.split(WorkflowExecutionContext.OUTPUT_FEED_SEPARATOR));
+ }
+
+ @Test
+ public void testGetOutputFeedInstancePaths() throws Exception {
+ Assert.assertEquals(context.getOutputFeedInstancePaths(), OUTPUT_INSTANCE_PATHS);
+ }
+
+ @Test
+ public void testGetOutputFeedInstancePathsList() throws Exception {
+ Assert.assertEquals(context.getOutputFeedInstancePathsList(),
+ OUTPUT_INSTANCE_PATHS.split(","));
+ }
+
+ @Test
+ public void testGetInputFeedNames() throws Exception {
+ Assert.assertEquals(context.getOutputFeedNames(), OUTPUT_FEED_NAMES);
+ }
+
+ @Test
+ public void testGetInputFeedNamesList() throws Exception {
+ Assert.assertEquals(context.getInputFeedNamesList(),
+ INPUT_FEED_NAMES.split(WorkflowExecutionContext.INPUT_FEED_SEPARATOR));
+ }
+
+ @Test
+ public void testGetInputFeedInstancePaths() throws Exception {
+ Assert.assertEquals(context.getInputFeedInstancePaths(), INPUT_INSTANCE_PATHS);
+ }
+
+ @Test
+ public void testGetInputFeedInstancePathsList() throws Exception {
+ Assert.assertEquals(context.getInputFeedInstancePathsList(),
+ INPUT_INSTANCE_PATHS.split("#"));
+ }
+
+ @Test
+ public void testGetWorkflowEngineUrl() throws Exception {
+ Assert.assertEquals(context.getWorkflowEngineUrl(), "http://localhost:11000/oozie");
+ }
+
+ @Test
+ public void testGetUserWorkflowEngine() throws Exception {
+ Assert.assertEquals(context.getUserWorkflowEngine(), EngineType.PIG.name());
+ }
+
+ @Test
+ public void testGetUserWorkflowVersion() throws Exception {
+ Assert.assertEquals(context.getUserWorkflowVersion(), WORKFLOW_VERSION);
+ }
+
+ @Test
+ public void testGetWorkflowId() throws Exception {
+ Assert.assertEquals(context.getWorkflowId(), "workflow-01-00");
+ }
+
+ @Test
+ public void testGetUserSubflowId() throws Exception {
+ Assert.assertEquals(context.getUserSubflowId(), "userflow@wf-id");
+ }
+
+ @Test
+ public void testGetWorkflowRunId() throws Exception {
+ Assert.assertEquals(context.getWorkflowRunId(), 1);
+ }
+
+ @Test
+ public void testGetWorkflowRunIdString() throws Exception {
+ Assert.assertEquals(context.getWorkflowRunIdString(), "1");
+ }
+
+ @Test
+ public void testGetWorkflowUser() throws Exception {
+ Assert.assertEquals(context.getWorkflowUser(), FALCON_USER);
+ }
+
+ @Test
+ public void testGetExecutionCompletionTime() throws Exception {
+ Assert.assertNotNull(context.getExecutionCompletionTime());
+ }
+
+ @Test
+ public void testSerializeDeserialize() throws Exception {
+ String contextFile = context.getContextFile();
+ context.serialize();
+ WorkflowExecutionContext newContext = WorkflowExecutionContext.deSerialize(contextFile);
+ Assert.assertNotNull(newContext);
+ Assert.assertEquals(newContext.entrySet().size(), context.entrySet().size());
+ }
+
+ @Test
+ public void testSerializeDeserializeWithFile() throws Exception {
+ String contextFile = "/tmp/blah.json";
+ context.serialize(contextFile);
+ WorkflowExecutionContext newContext = WorkflowExecutionContext.deSerialize(contextFile);
+ Assert.assertNotNull(newContext);
+ Assert.assertEquals(newContext.entrySet().size(), context.entrySet().size());
+ }
+
+ @Test
+ public void testGetFilePath() throws Exception {
+ Assert.assertEquals(WorkflowExecutionContext.getFilePath(LOGS_DIR, ENTITY_NAME),
+ LOGS_DIR + "/" + ENTITY_NAME + "-wf-post-exec-context.json");
+ }
+
+ private static String[] getTestMessageArgs() {
+ return new String[]{
+ "-" + WorkflowExecutionArgs.CLUSTER_NAME.getName(), CLUSTER_NAME,
+ "-" + WorkflowExecutionArgs.ENTITY_TYPE.getName(), "process",
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), ENTITY_NAME,
+ "-" + WorkflowExecutionArgs.NOMINAL_TIME.getName(), NOMINAL_TIME,
+ "-" + WorkflowExecutionArgs.OPERATION.getName(), OPERATION,
+
+ "-" + WorkflowExecutionArgs.INPUT_FEED_NAMES.getName(), INPUT_FEED_NAMES,
+ "-" + WorkflowExecutionArgs.INPUT_FEED_PATHS.getName(), INPUT_INSTANCE_PATHS,
+
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), OUTPUT_FEED_NAMES,
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(), OUTPUT_INSTANCE_PATHS,
+
+ "-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
+ "-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), FALCON_USER,
+ "-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
+ "-" + WorkflowExecutionArgs.STATUS.getName(), "SUCCEEDED",
+ "-" + WorkflowExecutionArgs.TIMESTAMP.getName(), NOMINAL_TIME,
+
+ "-" + WorkflowExecutionArgs.WF_ENGINE_URL.getName(), "http://localhost:11000/oozie",
+ "-" + WorkflowExecutionArgs.USER_SUBFLOW_ID.getName(), "userflow@wf-id",
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_NAME.getName(), WORKFLOW_NAME,
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_VERSION.getName(), WORKFLOW_VERSION,
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_ENGINE.getName(), EngineType.PIG.name(),
+
+ "-" + WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), BROKER,
+ "-" + WorkflowExecutionArgs.BRKR_URL.getName(), "tcp://localhost:61616?daemon=true",
+ "-" + WorkflowExecutionArgs.USER_BRKR_IMPL_CLASS.getName(), BROKER,
+ "-" + WorkflowExecutionArgs.USER_BRKR_URL.getName(), "tcp://localhost:61616?daemon=true",
+ "-" + WorkflowExecutionArgs.BRKR_TTL.getName(), "1000",
+
+ "-" + WorkflowExecutionArgs.LOG_DIR.getName(), LOGS_DIR,
+ "-" + WorkflowExecutionArgs.LOG_FILE.getName(), LOGS_DIR + "/log.txt",
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b42c53ba/src/conf/startup.properties
----------------------------------------------------------------------
diff --git a/src/conf/startup.properties b/src/conf/startup.properties
index 2c0bbbc..563c4b9 100644
--- a/src/conf/startup.properties
+++ b/src/conf/startup.properties
@@ -33,6 +33,7 @@
*.catalog.service.impl=org.apache.falcon.catalog.HiveCatalogService
*.application.services=org.apache.falcon.security.AuthenticationInitializationService,\
+ org.apache.falcon.workflow.WorkflowJobEndNotificationService, \
org.apache.falcon.service.ProcessSubscriberService,\
org.apache.falcon.entity.store.ConfigurationStore,\
org.apache.falcon.rerun.service.RetryService,\
[4/8] git commit: FALCON-486 - Introduce Workflow Context in Post
Processing. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-486 - Introduce Workflow Context in Post Processing. 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/c6000363
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/c6000363
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/c6000363
Branch: refs/heads/master
Commit: c6000363a19e11745d9f3f7abc14e03f08be7f88
Parents: 23762e5
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Fri Aug 8 12:58:23 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Fri Aug 8 12:58:23 2014 -0700
----------------------------------------------------------------------
.../org/apache/falcon/metadata/LineageArgs.java | 1 +
.../apache/falcon/metadata/LineageRecorder.java | 1 +
.../falcon/messaging/JMSMessageProducer.java | 57 ++---
.../org/apache/falcon/logging/JobLogMover.java | 170 ++++++++++++++
.../org/apache/falcon/logging/LogMover.java | 229 -------------------
.../falcon/workflow/FalconPostProcessing.java | 227 +++---------------
.../workflow/FalconPostProcessingTest.java | 78 +++----
.../org/apache/falcon/process/PigProcessIT.java | 2 +-
.../org/apache/falcon/util/OozieTestUtils.java | 15 +-
9 files changed, 268 insertions(+), 512 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/c6000363/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java b/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
index 7adf5bb..e91f7ab 100644
--- a/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
+++ b/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
@@ -24,6 +24,7 @@ import org.apache.commons.cli.Option;
/**
* Args for data Lineage.
*/
+@Deprecated // delete this class
public enum LineageArgs {
// process instance
NOMINAL_TIME("nominalTime", "instance time"),
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/c6000363/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java b/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
index 8a946ad..5ab1bf0 100644
--- a/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
+++ b/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
@@ -46,6 +46,7 @@ import java.util.Map;
/**
* Utility called in the post process of oozie workflow to record lineage information.
*/
+@Deprecated // delete this class
public class LineageRecorder extends Configured implements Tool {
private static final Logger LOG = LoggerFactory.getLogger(LineageRecorder.class);
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/c6000363/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageProducer.java
----------------------------------------------------------------------
diff --git a/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageProducer.java b/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageProducer.java
index 6c9859c..aeadf5c 100644
--- a/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageProducer.java
+++ b/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageProducer.java
@@ -63,35 +63,6 @@ public class JMSMessageProducer {
private static final long DEFAULT_TTL = 3 * 24 * 60 * 60 * 1000;
- private static final WorkflowExecutionArgs[] FALCON_FILTER = {
- WorkflowExecutionArgs.NOMINAL_TIME,
- WorkflowExecutionArgs.ENTITY_NAME,
- WorkflowExecutionArgs.OPERATION,
- WorkflowExecutionArgs.LOG_DIR,
- WorkflowExecutionArgs.STATUS,
- WorkflowExecutionArgs.CONTEXT_FILE,
- WorkflowExecutionArgs.TIMESTAMP,
- };
-
- private static final WorkflowExecutionArgs[] USER_FILTER = {
- WorkflowExecutionArgs.CLUSTER_NAME,
- WorkflowExecutionArgs.ENTITY_NAME,
- WorkflowExecutionArgs.ENTITY_TYPE,
- WorkflowExecutionArgs.NOMINAL_TIME,
- WorkflowExecutionArgs.OPERATION,
-
- WorkflowExecutionArgs.FEED_NAMES,
- WorkflowExecutionArgs.FEED_INSTANCE_PATHS,
-
- WorkflowExecutionArgs.WORKFLOW_ID,
- WorkflowExecutionArgs.WORKFLOW_USER,
- WorkflowExecutionArgs.RUN_ID,
- WorkflowExecutionArgs.STATUS,
- WorkflowExecutionArgs.TIMESTAMP,
- WorkflowExecutionArgs.LOG_FILE,
- };
-
-
private final WorkflowExecutionContext context;
private final MessageType messageType;
@@ -170,9 +141,26 @@ public class JMSMessageProducer {
* Accepts a Message to be send to JMS topic, creates a new
* Topic based on topic name if it does not exist or else
* existing topic with the same name is used to send the message.
+ * Sends all arguments.
+ *
+ * @return error code
+ * @throws JMSException
*/
public int sendMessage() throws JMSException {
- List<Map<String, String>> messageList = buildMessageList();
+ return sendMessage(WorkflowExecutionArgs.values());
+ }
+
+ /**
+ * Accepts a Message to be send to JMS topic, creates a new
+ * Topic based on topic name if it does not exist or else
+ * existing topic with the same name is used to send the message.
+ *
+ * @param filteredArgs args sent in the message.
+ * @return error code
+ * @throws JMSException
+ */
+ public int sendMessage(WorkflowExecutionArgs[] filteredArgs) throws JMSException {
+ List<Map<String, String>> messageList = buildMessageList(filteredArgs);
if (messageList.isEmpty()) {
LOG.warn("No operation on output feed");
@@ -198,10 +186,7 @@ public class JMSMessageProducer {
return 0;
}
- private List<Map<String, String>> buildMessageList() {
- WorkflowExecutionArgs[] fileredArgs = messageType == MessageType.FALCON
- ? FALCON_FILTER : USER_FILTER;
-
+ private List<Map<String, String>> buildMessageList(WorkflowExecutionArgs[] filteredArgs) {
String[] feedNames = getFeedNames();
if (feedNames == null) {
return Collections.emptyList();
@@ -217,7 +202,7 @@ public class JMSMessageProducer {
List<Map<String, String>> messages = new ArrayList<Map<String, String>>(feedPaths.length);
for (int i = 0; i < feedPaths.length; i++) {
- Map<String, String> message = buildMessage(fileredArgs);
+ Map<String, String> message = buildMessage(filteredArgs);
// override default values
if (context.getEntityType().equalsIgnoreCase("PROCESS")) {
@@ -347,7 +332,7 @@ public class JMSMessageProducer {
@SuppressWarnings("unchecked")
private Connection createAndStartConnection(String implementation, String userName,
- String password, String url)
+ String password, String url)
throws JMSException, ClassNotFoundException, InstantiationException,
IllegalAccessException, InvocationTargetException, NoSuchMethodException {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/c6000363/oozie/src/main/java/org/apache/falcon/logging/JobLogMover.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/logging/JobLogMover.java b/oozie/src/main/java/org/apache/falcon/logging/JobLogMover.java
new file mode 100644
index 0000000..1a08ada
--- /dev/null
+++ b/oozie/src/main/java/org/apache/falcon/logging/JobLogMover.java
@@ -0,0 +1,170 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.logging;
+
+import org.apache.falcon.entity.v0.EntityType;
+import org.apache.falcon.entity.v0.process.EngineType;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
+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.hadoop.util.ReflectionUtils;
+import org.apache.oozie.client.OozieClient;
+import org.apache.oozie.client.OozieClientException;
+import org.apache.oozie.client.WorkflowAction;
+import org.apache.oozie.client.WorkflowJob;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.URL;
+import java.net.URLConnection;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Utility called in the post process of oozie workflow to move oozie action executor log.
+ */
+public class JobLogMover {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JobLogMover.class);
+
+ public static final Set<String> FALCON_ACTIONS =
+ new HashSet<String>(Arrays.asList(new String[]{"eviction", "replication", }));
+
+ private Configuration getConf() {
+ return new Configuration();
+ }
+
+ public int run(WorkflowExecutionContext context) {
+ try {
+ OozieClient client = new OozieClient(context.getWorkflowEngineUrl());
+ WorkflowJob jobInfo;
+ try {
+ jobInfo = client.getJobInfo(context.getUserSubflowId());
+ } catch (OozieClientException e) {
+ LOG.error("Error getting jobinfo for: {}", context.getUserSubflowId(), e);
+ return 0;
+ }
+
+ Path path = new Path(context.getLogDir() + "/"
+ + String.format("%03d", context.getWorkflowRunId()));
+ FileSystem fs = path.getFileSystem(getConf());
+
+ if (EntityType.FEED.name().equalsIgnoreCase(context.getEntityType())
+ || notUserWorkflowEngineIsOozie(context.getUserWorkflowEngine())) {
+ // if replication wf, retention wf or PIG Process
+ copyOozieLog(client, fs, path, jobInfo.getId());
+
+ List<WorkflowAction> workflowActions = jobInfo.getActions();
+ for (int i=0; i < workflowActions.size(); i++) {
+ if (FALCON_ACTIONS.contains(workflowActions.get(i).getName())) {
+ copyTTlogs(fs, path, jobInfo.getActions().get(i));
+ break;
+ }
+ }
+ } else {
+ // if process wf with oozie engine
+ String subflowId = jobInfo.getExternalId();
+ copyOozieLog(client, fs, path, subflowId);
+ WorkflowJob subflowInfo = client.getJobInfo(subflowId);
+ List<WorkflowAction> actions = subflowInfo.getActions();
+ for (WorkflowAction action : actions) {
+ if (action.getType().equals("pig")
+ || action.getType().equals("java")) {
+ copyTTlogs(fs, path, action);
+ } else {
+ LOG.info("Ignoring hadoop TT log for non-pig and non-java action: {}", action.getName());
+ }
+ }
+ }
+
+ } catch (Exception e) {
+ // JobLogMover doesn't throw exception, a failed log mover will not fail the user workflow
+ LOG.error("Exception in log mover:", e);
+ }
+ return 0;
+ }
+
+ private boolean notUserWorkflowEngineIsOozie(String userWorkflowEngine) {
+ // userWorkflowEngine will be null for replication and "pig" for pig
+ return userWorkflowEngine != null && EngineType.fromValue(userWorkflowEngine) != EngineType.OOZIE;
+ }
+
+ private void copyOozieLog(OozieClient client, FileSystem fs, Path path,
+ String id) throws OozieClientException, IOException {
+ InputStream in = new ByteArrayInputStream(client.getJobLog(id).getBytes());
+ OutputStream out = fs.create(new Path(path, "oozie.log"));
+ IOUtils.copyBytes(in, out, 4096, true);
+ LOG.info("Copied oozie log to {}", path);
+ }
+
+ private void copyTTlogs(FileSystem fs, Path path,
+ WorkflowAction action) throws Exception {
+ String ttLogURL = getTTlogURL(action.getExternalId());
+ if (ttLogURL != null) {
+ LOG.info("Fetching log for action: {} from url: {}", action.getExternalId(), ttLogURL);
+ InputStream in = getURLinputStream(new URL(ttLogURL));
+ OutputStream out = fs.create(new Path(path, action.getName() + "_"
+ + getMappedStatus(action.getStatus()) + ".log"));
+ IOUtils.copyBytes(in, out, 4096, true);
+ LOG.info("Copied log to {}", path);
+ }
+ }
+
+ private String getMappedStatus(WorkflowAction.Status status) {
+ if (status == WorkflowAction.Status.FAILED
+ || status == WorkflowAction.Status.KILLED
+ || status == WorkflowAction.Status.ERROR) {
+ return "FAILED";
+ } else {
+ return "SUCCEEDED";
+ }
+ }
+
+ private String getTTlogURL(String jobId) throws Exception {
+ TaskLogURLRetriever logRetriever = ReflectionUtils
+ .newInstance(getLogRetrieverClassName(), getConf());
+ return logRetriever.retrieveTaskLogURL(jobId);
+ }
+
+ @SuppressWarnings("unchecked")
+ private Class<? extends TaskLogURLRetriever> getLogRetrieverClassName() {
+ try {
+ return (Class<? extends TaskLogURLRetriever>)
+ Class.forName("org.apache.falcon.logging.v1.TaskLogRetrieverV1");
+ } catch (ClassNotFoundException e) {
+ LOG.warn("V1 Retriever missing, falling back to Default retriever");
+ return DefaultTaskLogRetriever.class;
+ }
+ }
+
+ private InputStream getURLinputStream(URL url) throws IOException {
+ URLConnection connection = url.openConnection();
+ connection.setDoOutput(true);
+ connection.connect();
+ return connection.getInputStream();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/c6000363/oozie/src/main/java/org/apache/falcon/logging/LogMover.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/logging/LogMover.java b/oozie/src/main/java/org/apache/falcon/logging/LogMover.java
deleted file mode 100644
index 3922b38..0000000
--- a/oozie/src/main/java/org/apache/falcon/logging/LogMover.java
+++ /dev/null
@@ -1,229 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-package org.apache.falcon.logging;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.GnuParser;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.apache.falcon.entity.v0.EntityType;
-import org.apache.falcon.entity.v0.process.EngineType;
-import org.apache.hadoop.conf.Configured;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.util.ReflectionUtils;
-import org.apache.hadoop.util.Tool;
-import org.apache.hadoop.util.ToolRunner;
-import org.apache.oozie.client.OozieClient;
-import org.apache.oozie.client.OozieClientException;
-import org.apache.oozie.client.WorkflowAction;
-import org.apache.oozie.client.WorkflowJob;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.URL;
-import java.net.URLConnection;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-/**
- * Utility called in the post process of oozie workflow to move oozie action executor log.
- */
-public class LogMover extends Configured implements Tool {
-
- private static final Logger LOG = LoggerFactory.getLogger(LogMover.class);
- public static final Set<String> FALCON_ACTIONS =
- new HashSet<String>(Arrays.asList(new String[]{"eviction", "replication", }));
-
- /**
- * Args to the command.
- */
- private static class ARGS {
- private String oozieUrl;
- private String subflowId;
- private String runId;
- private String logDir;
- private String entityType;
- private String userWorkflowEngine;
- }
-
- public static void main(String[] args) throws Exception {
- ToolRunner.run(new LogMover(), args);
- }
-
- @Override
- public int run(String[] arguments) throws Exception {
- try {
- ARGS args = new ARGS();
- setupArgs(arguments, args);
-
- OozieClient client = new OozieClient(args.oozieUrl);
- WorkflowJob jobInfo;
- try {
- jobInfo = client.getJobInfo(args.subflowId);
- } catch (OozieClientException e) {
- LOG.error("Error getting jobinfo for: {}", args.subflowId, e);
- return 0;
- }
-
- Path path = new Path(args.logDir + "/"
- + String.format("%03d", Integer.parseInt(args.runId)));
- FileSystem fs = path.getFileSystem(getConf());
-
- if (args.entityType.equalsIgnoreCase(EntityType.FEED.name())
- || notUserWorkflowEngineIsOozie(args.userWorkflowEngine)) {
- // if replication wf, retention wf or PIG Process
- copyOozieLog(client, fs, path, jobInfo.getId());
-
- List<WorkflowAction> workflowActions = jobInfo.getActions();
- for (int i=0; i < workflowActions.size(); i++) {
- if (FALCON_ACTIONS.contains(workflowActions.get(i).getName())) {
- copyTTlogs(fs, path, jobInfo.getActions().get(i));
- break;
- }
- }
- } else {
- // if process wf with oozie engine
- String subflowId = jobInfo.getExternalId();
- copyOozieLog(client, fs, path, subflowId);
- WorkflowJob subflowInfo = client.getJobInfo(subflowId);
- List<WorkflowAction> actions = subflowInfo.getActions();
- for (WorkflowAction action : actions) {
- if (action.getType().equals("pig")
- || action.getType().equals("java")) {
- copyTTlogs(fs, path, action);
- } else {
- LOG.info("Ignoring hadoop TT log for non-pig and non-java action: {}", action.getName());
- }
- }
- }
-
- } catch (Exception e) {
- LOG.error("Exception in log mover", e);
- }
- return 0;
- }
-
- private boolean notUserWorkflowEngineIsOozie(String userWorkflowEngine) {
- // userWorkflowEngine will be null for replication and "pig" for pig
- return userWorkflowEngine != null && EngineType.fromValue(userWorkflowEngine) != EngineType.OOZIE;
- }
-
- private void copyOozieLog(OozieClient client, FileSystem fs, Path path,
- String id) throws OozieClientException, IOException {
- InputStream in = new ByteArrayInputStream(client.getJobLog(id).getBytes());
- OutputStream out = fs.create(new Path(path, "oozie.log"));
- IOUtils.copyBytes(in, out, 4096, true);
- LOG.info("Copied oozie log to {}", path);
- }
-
- private void copyTTlogs(FileSystem fs, Path path,
- WorkflowAction action) throws Exception {
- String ttLogURL = getTTlogURL(action.getExternalId());
- if (ttLogURL != null) {
- LOG.info("Fetching log for action: {} from url: {}", action.getExternalId(), ttLogURL);
- InputStream in = getURLinputStream(new URL(ttLogURL));
- OutputStream out = fs.create(new Path(path, action.getName() + "_"
- + getMappedStatus(action.getStatus()) + ".log"));
- IOUtils.copyBytes(in, out, 4096, true);
- LOG.info("Copied log to {}", path);
- }
- }
-
- private String getMappedStatus(WorkflowAction.Status status) {
- if (status == WorkflowAction.Status.FAILED
- || status == WorkflowAction.Status.KILLED
- || status == WorkflowAction.Status.ERROR) {
- return "FAILED";
- } else {
- return "SUCCEEDED";
- }
- }
-
- private void setupArgs(String[] arguments, ARGS args) throws ParseException {
- Options options = new Options();
-
- Option opt = new Option("workflowEngineUrl", true, "url of workflow engine, ex:oozie");
- opt.setRequired(true);
- options.addOption(opt);
-
- opt = new Option("subflowId", true, "external id of userworkflow");
- opt.setRequired(true);
- options.addOption(opt);
-
- opt = new Option("userWorkflowEngine", true, "user workflow engine type");
- opt.setRequired(false); // replication will NOT have this arg sent
- options.addOption(opt);
-
- opt = new Option("runId", true, "current workflow's runid");
- opt.setRequired(true);
- options.addOption(opt);
-
- opt = new Option("logDir", true, "log dir where job logs are stored");
- opt.setRequired(true);
- options.addOption(opt);
-
- opt = new Option("status", true, "user workflow status");
- opt.setRequired(true);
- options.addOption(opt);
-
- opt = new Option("entityType", true, "entity type feed or process");
- opt.setRequired(true);
- options.addOption(opt);
-
- CommandLine cmd = new GnuParser().parse(options, arguments);
-
- args.oozieUrl = cmd.getOptionValue("workflowEngineUrl");
- args.subflowId = cmd.getOptionValue("subflowId");
- args.userWorkflowEngine = cmd.getOptionValue("userWorkflowEngine");
- args.runId = cmd.getOptionValue("runId");
- args.logDir = cmd.getOptionValue("logDir");
- args.entityType = cmd.getOptionValue("entityType");
- }
-
- private String getTTlogURL(String jobId) throws Exception {
- TaskLogURLRetriever logRetriever = ReflectionUtils.newInstance(getLogRetrieverClassName(), getConf());
- return logRetriever.retrieveTaskLogURL(jobId);
- }
-
- @SuppressWarnings("unchecked")
- private Class<? extends TaskLogURLRetriever> getLogRetrieverClassName() {
- try {
- return (Class<? extends TaskLogURLRetriever>)
- Class.forName("org.apache.falcon.logging.v1.TaskLogRetrieverV1");
- } catch (ClassNotFoundException e) {
- LOG.warn("V1 Retriever missing, falling back to Default retriever");
- return DefaultTaskLogRetriever.class;
- }
- }
-
- private InputStream getURLinputStream(URL url) throws IOException {
- URLConnection connection = url.openConnection();
- connection.setDoOutput(true);
- connection.connect();
- return connection.getInputStream();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/c6000363/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java b/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
index d3befa2..0adc11b 100644
--- a/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
+++ b/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
@@ -17,12 +17,8 @@
*/
package org.apache.falcon.workflow;
-import org.apache.commons.cli.*;
-import org.apache.commons.lang.StringUtils;
-import org.apache.falcon.logging.LogMover;
-import org.apache.falcon.messaging.MessageProducer;
-import org.apache.falcon.metadata.LineageArgs;
-import org.apache.falcon.metadata.LineageRecorder;
+import org.apache.falcon.logging.JobLogMover;
+import org.apache.falcon.messaging.JMSMessageProducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.security.UserGroupInformation;
@@ -31,68 +27,12 @@ import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
-import java.util.List;
-
/**
* Utility called by oozie workflow engine post workflow execution in parent workflow.
*/
public class FalconPostProcessing extends Configured implements Tool {
private static final Logger LOG = LoggerFactory.getLogger(FalconPostProcessing.class);
- /**
- * Args that the utility understands.
- */
- public enum Arg {
- CLUSTER("cluster", "name of the current cluster"),
- ENTITY_TYPE("entityType", "type of the entity"),
- ENTITY_NAME("entityName", "name of the entity"),
- NOMINAL_TIME("nominalTime", "instance time"),
- OPERATION("operation", "operation like generate, delete, replicate"),
- WORKFLOW_ID("workflowId", "current workflow-id of the instance"),
- RUN_ID("runId", "current run-id of the instance"),
- STATUS("status", "status of the user workflow instance"),
- TIMESTAMP("timeStamp", "current timestamp"),
- TOPIC_NAME("topicName", "name of the topic to be used to send JMS message"),
- BRKR_IMPL_CLASS("brokerImplClass", "falcon message broker Implementation class"),
- BRKR_URL("brokerUrl", "falcon message broker url"),
- USER_BRKR_IMPL_CLASS("userBrokerImplClass", "user broker Impl class"),
- USER_BRKR_URL("userBrokerUrl", "user broker url"),
- BRKR_TTL("brokerTTL", "time to live for broker message in sec"),
- FEED_NAMES("feedNames", "name of the feeds which are generated/replicated/deleted"),
- FEED_INSTANCE_PATHS("feedInstancePaths", "comma separated feed instance paths"),
- LOG_FILE("logFile", "log file path where feeds to be deleted are recorded"),
- WF_ENGINE_URL("workflowEngineUrl", "url of workflow engine server, ex:oozie"),
- USER_SUBFLOW_ID("subflowId", "external id of user workflow"),
- USER_WORKFLOW_ENGINE("userWorkflowEngine", "user workflow engine type"),
- USER_WORKFLOW_NAME("userWorkflowName", "user workflow name"),
- USER_WORKFLOW_VERSION("userWorkflowVersion", "user workflow version"),
- LOG_DIR("logDir", "log dir where job logs are copied"),
- WORKFLOW_USER("workflowUser", "user who owns the feed instance (partition)"),
- INPUT_FEED_NAMES("falconInputFeeds", "name of the feeds which are used as inputs"),
- INPUT_FEED_PATHS("falconInputPaths", "comma separated input feed instance paths");
-
- private String name;
- private String description;
-
- Arg(String name, String description) {
- this.name = name;
- this.description = description;
- }
-
- public Option getOption() {
- return new Option(this.name, true, this.description);
- }
-
- public String getOptionName() {
- return this.name;
- }
-
- public String getOptionValue(CommandLine cmd) {
- return cmd.getOptionValue(this.name);
- }
- }
-
public static void main(String[] args) throws Exception {
ToolRunner.run(new Configuration(), new FalconPostProcessing(), args);
}
@@ -100,161 +40,46 @@ public class FalconPostProcessing extends Configured implements Tool {
@Override
public int run(String[] args) throws Exception {
- CommandLine cmd = getCommand(args);
-
- LOG.info("Sending user message {}", cmd);
- invokeUserMessageProducer(cmd);
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(args,
+ WorkflowExecutionContext.Type.POST_PROCESSING);
+ LOG.info("Post workflow execution context created {}", context);
+ // serialize the context to HDFS under logs dir before sending the message
+ context.serialize();
- if ("SUCCEEDED".equals(Arg.STATUS.getOptionValue(cmd))) {
- LOG.info("Recording lineage for {}", cmd);
- recordLineageMetadata(cmd);
- }
+ LOG.info("Sending user message {} ", context);
+ invokeUserMessageProducer(context);
- //LogMover doesn't throw exception, a failed log mover will not fail the user workflow
- LOG.info("Moving logs {}", cmd);
- invokeLogProducer(cmd);
+ // JobLogMover doesn't throw exception, a failed log mover will not fail the user workflow
+ LOG.info("Moving logs {}", context);
+ invokeLogProducer(context);
- LOG.info("Sending falcon message {}", cmd);
- invokeFalconMessageProducer(cmd);
+ LOG.info("Sending falcon message {}", context);
+ invokeFalconMessageProducer(context);
return 0;
}
- private void invokeUserMessageProducer(CommandLine cmd) throws Exception {
- List<String> args = new ArrayList<String>();
- addArg(args, cmd, Arg.CLUSTER);
- addArg(args, cmd, Arg.ENTITY_TYPE);
- addArg(args, cmd, Arg.ENTITY_NAME);
- addArg(args, cmd, Arg.NOMINAL_TIME);
- addArg(args, cmd, Arg.OPERATION);
- addArg(args, cmd, Arg.WORKFLOW_ID);
- addArg(args, cmd, Arg.RUN_ID);
- addArg(args, cmd, Arg.STATUS);
- addArg(args, cmd, Arg.TIMESTAMP);
- //special args for user JMS message producer
- args.add("-" + Arg.TOPIC_NAME.getOptionName()); //user topic
- args.add("FALCON." + Arg.ENTITY_NAME.getOptionValue(cmd));
- //note, the user broker impl class arg name to MessageProducer is brokerImplClass
- args.add("-" + Arg.BRKR_IMPL_CLASS.getOptionName());
- args.add(Arg.USER_BRKR_IMPL_CLASS.getOptionValue(cmd));
- args.add("-" + Arg.BRKR_URL.getOptionName());
- args.add(Arg.USER_BRKR_URL.getOptionValue(cmd));
- addArg(args, cmd, Arg.BRKR_TTL);
- addArg(args, cmd, Arg.FEED_NAMES);
- addArg(args, cmd, Arg.FEED_INSTANCE_PATHS);
- addArg(args, cmd, Arg.LOG_FILE);
-
- MessageProducer.main(args.toArray(new String[0]));
+ private void invokeUserMessageProducer(WorkflowExecutionContext context) throws Exception {
+ JMSMessageProducer jmsMessageProducer = JMSMessageProducer.builder(context)
+ .type(JMSMessageProducer.MessageType.USER)
+ .build();
+ jmsMessageProducer.sendMessage(WorkflowExecutionContext.USER_MESSAGE_ARGS);
}
- private void invokeFalconMessageProducer(CommandLine cmd) throws Exception {
- List<String> args = new ArrayList<String>();
- addArg(args, cmd, Arg.CLUSTER);
- addArg(args, cmd, Arg.ENTITY_TYPE);
- addArg(args, cmd, Arg.ENTITY_NAME);
- addArg(args, cmd, Arg.NOMINAL_TIME);
- addArg(args, cmd, Arg.OPERATION);
- addArg(args, cmd, Arg.WORKFLOW_ID);
- addArg(args, cmd, Arg.RUN_ID);
- addArg(args, cmd, Arg.STATUS);
- addArg(args, cmd, Arg.TIMESTAMP);
- //special args Falcon JMS message producer
- args.add("-" + Arg.TOPIC_NAME.getOptionName());
- args.add("FALCON.ENTITY.TOPIC");
- args.add("-" + Arg.BRKR_IMPL_CLASS.getOptionName());
- args.add(Arg.BRKR_IMPL_CLASS.getOptionValue(cmd));
- args.add("-" + Arg.BRKR_URL.getOptionName());
- args.add(Arg.BRKR_URL.getOptionValue(cmd));
- addArg(args, cmd, Arg.BRKR_TTL);
- addArg(args, cmd, Arg.FEED_NAMES);
- addArg(args, cmd, Arg.FEED_INSTANCE_PATHS);
- addArg(args, cmd, Arg.LOG_FILE);
- addArg(args, cmd, Arg.WORKFLOW_USER);
- addArg(args, cmd, Arg.LOG_DIR);
-
- MessageProducer.main(args.toArray(new String[0]));
+ private void invokeFalconMessageProducer(WorkflowExecutionContext context) throws Exception {
+ JMSMessageProducer jmsMessageProducer = JMSMessageProducer.builder(context)
+ .type(JMSMessageProducer.MessageType.FALCON)
+ .build();
+ jmsMessageProducer.sendMessage();
}
- private void invokeLogProducer(CommandLine cmd) throws Exception {
+ private void invokeLogProducer(WorkflowExecutionContext context) {
// todo: need to move this out to Falcon in-process
if (UserGroupInformation.isSecurityEnabled()) {
LOG.info("Unable to move logs as security is enabled.");
return;
}
- List<String> args = new ArrayList<String>();
- addArg(args, cmd, Arg.WF_ENGINE_URL);
- addArg(args, cmd, Arg.ENTITY_TYPE);
- addArg(args, cmd, Arg.USER_SUBFLOW_ID);
- addArg(args, cmd, Arg.USER_WORKFLOW_ENGINE);
- addArg(args, cmd, Arg.RUN_ID);
- addArg(args, cmd, Arg.LOG_DIR);
- addArg(args, cmd, Arg.STATUS);
-
- LogMover.main(args.toArray(new String[0]));
- }
-
- private void recordLineageMetadata(CommandLine cmd) throws Exception {
- List<String> args = new ArrayList<String>();
-
- for (LineageArgs arg : LineageArgs.values()) {
- if (StringUtils.isNotEmpty(arg.getOptionValue(cmd))) {
- args.add("-" + arg.getOptionName());
- args.add(arg.getOptionValue(cmd));
- }
- }
-
- LineageRecorder.main(args.toArray(new String[args.size()]));
- }
-
- private void addArg(List<String> args, CommandLine cmd, Arg arg) {
- if (StringUtils.isNotEmpty(arg.getOptionValue(cmd))) {
- args.add("-" + arg.getOptionName());
- args.add(arg.getOptionValue(cmd));
- }
- }
-
- private static CommandLine getCommand(String[] arguments)
- throws ParseException {
-
- Options options = new Options();
- addOption(options, Arg.CLUSTER);
- addOption(options, Arg.ENTITY_TYPE);
- addOption(options, Arg.ENTITY_NAME);
- addOption(options, Arg.NOMINAL_TIME);
- addOption(options, Arg.OPERATION);
- addOption(options, Arg.WORKFLOW_ID);
- addOption(options, Arg.RUN_ID);
- addOption(options, Arg.STATUS);
- addOption(options, Arg.TIMESTAMP);
- addOption(options, Arg.BRKR_IMPL_CLASS);
- addOption(options, Arg.BRKR_URL);
- addOption(options, Arg.USER_BRKR_IMPL_CLASS);
- addOption(options, Arg.USER_BRKR_URL);
- addOption(options, Arg.BRKR_TTL);
- addOption(options, Arg.FEED_NAMES);
- addOption(options, Arg.FEED_INSTANCE_PATHS);
- addOption(options, Arg.LOG_FILE);
- addOption(options, Arg.WF_ENGINE_URL);
- addOption(options, Arg.USER_SUBFLOW_ID);
- addOption(options, Arg.USER_WORKFLOW_NAME, false);
- addOption(options, Arg.USER_WORKFLOW_VERSION, false);
- addOption(options, Arg.USER_WORKFLOW_ENGINE, false);
- addOption(options, Arg.LOG_DIR);
- addOption(options, Arg.WORKFLOW_USER);
- addOption(options, Arg.INPUT_FEED_NAMES, false);
- addOption(options, Arg.INPUT_FEED_PATHS, false);
-
- return new GnuParser().parse(options, arguments);
- }
-
- private static void addOption(Options options, Arg arg) {
- addOption(options, arg, true);
- }
-
- private static void addOption(Options options, Arg arg, boolean isRequired) {
- Option option = arg.getOption();
- option.setRequired(isRequired);
- options.addOption(option);
+ new JobLogMover().run(context);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/c6000363/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 561303d..feddfdd 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
@@ -20,7 +20,7 @@ package org.apache.falcon.oozie.workflow;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.falcon.workflow.FalconPostProcessing;
-import org.apache.falcon.workflow.FalconPostProcessing.Arg;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -45,34 +45,34 @@ public class FalconPostProcessingTest {
@BeforeClass
public void setup() throws Exception {
args = new String[]{
- "-" + Arg.ENTITY_NAME.getOptionName(), ENTITY_NAME,
- "-" + Arg.FEED_NAMES.getOptionName(), "out-click-logs,out-raw-logs",
- "-" + Arg.FEED_INSTANCE_PATHS.getOptionName(),
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), ENTITY_NAME,
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), "out-click-logs,out-raw-logs",
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(),
"/out-click-logs/10/05/05/00/20,/out-raw-logs/10/05/05/00/20",
- "-" + Arg.WORKFLOW_ID.getOptionName(), "workflow-01-00",
- "-" + Arg.WORKFLOW_USER.getOptionName(), "falcon",
- "-" + Arg.RUN_ID.getOptionName(), "1",
- "-" + Arg.NOMINAL_TIME.getOptionName(), "2011-01-01-01-00",
- "-" + Arg.TIMESTAMP.getOptionName(), "2012-01-01-01-00",
- "-" + Arg.BRKR_URL.getOptionName(), BROKER_URL,
- "-" + Arg.BRKR_IMPL_CLASS.getOptionName(), (BROKER_IMPL_CLASS),
- "-" + Arg.USER_BRKR_URL.getOptionName(), BROKER_URL,
- "-" + Arg.USER_BRKR_IMPL_CLASS.getOptionName(), (BROKER_IMPL_CLASS),
- "-" + Arg.ENTITY_TYPE.getOptionName(), ("process"),
- "-" + Arg.OPERATION.getOptionName(), ("GENERATE"),
- "-" + Arg.LOG_FILE.getOptionName(), ("/logFile"),
- "-" + Arg.STATUS.getOptionName(), ("SUCCEEDED"),
- "-" + Arg.BRKR_TTL.getOptionName(), "10",
- "-" + Arg.CLUSTER.getOptionName(), "corp",
- "-" + Arg.WF_ENGINE_URL.getOptionName(), "http://localhost:11000/oozie/",
- "-" + Arg.LOG_DIR.getOptionName(), "target/log",
- "-" + Arg.USER_SUBFLOW_ID.getOptionName(), "userflow@wf-id" + "test",
- "-" + Arg.USER_WORKFLOW_ENGINE.getOptionName(), "oozie",
- "-" + Arg.INPUT_FEED_NAMES.getOptionName(), "in-click-logs,in-raw-logs",
- "-" + Arg.INPUT_FEED_PATHS.getOptionName(),
+ "-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
+ "-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), "falcon",
+ "-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
+ "-" + WorkflowExecutionArgs.NOMINAL_TIME.getName(), "2011-01-01-01-00",
+ "-" + WorkflowExecutionArgs.TIMESTAMP.getName(), "2012-01-01-01-00",
+ "-" + WorkflowExecutionArgs.BRKR_URL.getName(), BROKER_URL,
+ "-" + WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), BROKER_IMPL_CLASS,
+ "-" + WorkflowExecutionArgs.USER_BRKR_URL.getName(), BROKER_URL,
+ "-" + WorkflowExecutionArgs.USER_BRKR_IMPL_CLASS.getName(), BROKER_IMPL_CLASS,
+ "-" + WorkflowExecutionArgs.ENTITY_TYPE.getName(), "process",
+ "-" + WorkflowExecutionArgs.OPERATION.getName(), "GENERATE",
+ "-" + WorkflowExecutionArgs.LOG_FILE.getName(), "/logFile",
+ "-" + WorkflowExecutionArgs.STATUS.getName(), "SUCCEEDED",
+ "-" + WorkflowExecutionArgs.BRKR_TTL.getName(), "10",
+ "-" + WorkflowExecutionArgs.CLUSTER_NAME.getName(), "corp",
+ "-" + WorkflowExecutionArgs.WF_ENGINE_URL.getName(), "http://localhost:11000/oozie/",
+ "-" + WorkflowExecutionArgs.LOG_DIR.getName(), "target/log",
+ "-" + WorkflowExecutionArgs.USER_SUBFLOW_ID.getName(), "userflow@wf-id" + "test",
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_ENGINE.getName(), "oozie",
+ "-" + WorkflowExecutionArgs.INPUT_FEED_NAMES.getName(), "in-click-logs,in-raw-logs",
+ "-" + WorkflowExecutionArgs.INPUT_FEED_PATHS.getName(),
"/in-click-logs/10/05/05/00/20,/in-raw-logs/10/05/05/00/20",
- "-" + Arg.USER_WORKFLOW_NAME.getOptionName(), "test-workflow",
- "-" + Arg.USER_WORKFLOW_VERSION.getOptionName(), "1.0.0",
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_NAME.getName(), "test-workflow",
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_VERSION.getName(), "1.0.0",
};
broker = new BrokerService();
@@ -95,8 +95,8 @@ public class FalconPostProcessingTest {
@Override
public void run() {
try {
- consumer(BROKER_URL, "FALCON." + ENTITY_NAME);
- consumer(BROKER_URL, FALCON_TOPIC_NAME);
+ consumer(BROKER_URL, "FALCON." + ENTITY_NAME); // user message
+ consumer(BROKER_URL, FALCON_TOPIC_NAME); // falcon message
} catch (AssertionError e) {
error = e;
} catch (JMSException ignore) {
@@ -131,13 +131,13 @@ public class FalconPostProcessingTest {
assertMessage(m);
if (topic.equals(FALCON_TOPIC_NAME)) {
- Assert.assertEquals(m.getString(Arg.FEED_NAMES.getOptionName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_NAMES.getName()),
"out-click-logs,out-raw-logs");
- Assert.assertEquals(m.getString(Arg.FEED_INSTANCE_PATHS.getOptionName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName()),
"/out-click-logs/10/05/05/00/20,/out-raw-logs/10/05/05/00/20");
} else {
- Assert.assertEquals(m.getString(Arg.FEED_NAMES.getOptionName()), "out-click-logs");
- Assert.assertEquals(m.getString(Arg.FEED_INSTANCE_PATHS.getOptionName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_NAMES.getName()), "out-click-logs");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName()),
"/out-click-logs/10/05/05/00/20");
}
@@ -145,15 +145,13 @@ public class FalconPostProcessingTest {
}
private void assertMessage(MapMessage m) throws JMSException {
- Assert.assertEquals(m.getString(Arg.ENTITY_NAME.getOptionName()), "agg-coord");
- Assert.assertEquals(m.getString(Arg.WORKFLOW_ID.getOptionName()), "workflow-01-00");
- String workflowUser = m.getString(Arg.WORKFLOW_USER.getOptionName());
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.ENTITY_NAME.getName()), "agg-coord");
+ String workflowUser = m.getString(WorkflowExecutionArgs.WORKFLOW_USER.getName());
if (workflowUser != null) { // in case of user message, its NULL
Assert.assertEquals(workflowUser, "falcon");
}
- Assert.assertEquals(m.getString(Arg.RUN_ID.getOptionName()), "1");
- Assert.assertEquals(m.getString(Arg.NOMINAL_TIME.getOptionName()), "2011-01-01T01:00Z");
- Assert.assertEquals(m.getString(Arg.TIMESTAMP.getOptionName()), "2012-01-01T01:00Z");
- Assert.assertEquals(m.getString(Arg.STATUS.getOptionName()), "SUCCEEDED");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.NOMINAL_TIME.getName()), "2011-01-01T01:00Z");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.TIMESTAMP.getName()), "2012-01-01T01:00Z");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.STATUS.getName()), "SUCCEEDED");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/c6000363/webapp/src/test/java/org/apache/falcon/process/PigProcessIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/falcon/process/PigProcessIT.java b/webapp/src/test/java/org/apache/falcon/process/PigProcessIT.java
index 51def35..89132d6 100644
--- a/webapp/src/test/java/org/apache/falcon/process/PigProcessIT.java
+++ b/webapp/src/test/java/org/apache/falcon/process/PigProcessIT.java
@@ -117,7 +117,7 @@ public class PigProcessIT {
.get(InstancesResult.class);
Assert.assertEquals(APIResult.Status.SUCCEEDED, response.getStatus());
- // verify LogMover
+ // verify JobLogMover
Path oozieLogPath = OozieTestUtils.getOozieLogPath(context.getCluster().getCluster(), jobInfo);
Assert.assertTrue(context.getCluster().getFileSystem().exists(oozieLogPath));
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/c6000363/webapp/src/test/java/org/apache/falcon/util/OozieTestUtils.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/falcon/util/OozieTestUtils.java b/webapp/src/test/java/org/apache/falcon/util/OozieTestUtils.java
index de8bc33..e67fe2a 100644
--- a/webapp/src/test/java/org/apache/falcon/util/OozieTestUtils.java
+++ b/webapp/src/test/java/org/apache/falcon/util/OozieTestUtils.java
@@ -22,8 +22,9 @@ import org.apache.falcon.FalconException;
import org.apache.falcon.entity.ClusterHelper;
import org.apache.falcon.entity.EntityUtil;
import org.apache.falcon.entity.v0.cluster.Cluster;
-import org.apache.falcon.logging.LogMover;
+import org.apache.falcon.logging.JobLogMover;
import org.apache.falcon.resource.TestContext;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
import org.apache.falcon.workflow.engine.OozieClientFactory;
import org.apache.hadoop.fs.Path;
import org.apache.oozie.client.BundleJob;
@@ -176,13 +177,17 @@ public final class OozieTestUtils {
public static Path getOozieLogPath(Cluster cluster, WorkflowJob jobInfo) throws Exception {
Path stagingPath = EntityUtil.getLogPath(cluster, cluster);
final Path logPath = new Path(ClusterHelper.getStorageUrl(cluster), stagingPath);
- LogMover.main(new String[] {
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(new String[] {
"-workflowEngineUrl", ClusterHelper.getOozieUrl(cluster),
- "-subflowId", jobInfo.getId(), "-runId", "1",
+ "-subflowId", jobInfo.getId(),
+ "-runId", "1",
"-logDir", logPath.toString() + "/job-2012-04-21-00-00",
- "-status", "SUCCEEDED", "-entityType", "process",
+ "-status", "SUCCEEDED",
+ "-entityType", "process",
"-userWorkflowEngine", "pig",
- });
+ }, WorkflowExecutionContext.Type.POST_PROCESSING);
+
+ new JobLogMover().run(context);
return new Path(logPath, "job-2012-04-21-00-00/001/oozie.log");
}
[2/8] FALCON-485 - Simplify JMS Message Sender/Consumer and use
Workflow Context. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/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 9f8b07f..95a3780 100644
--- a/messaging/src/test/java/org/apache/falcon/messaging/ProcessProducerTest.java
+++ b/messaging/src/test/java/org/apache/falcon/messaging/ProcessProducerTest.java
@@ -20,13 +20,20 @@ package org.apache.falcon.messaging;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-import javax.jms.*;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
/**
* Test for process message producer.
@@ -36,31 +43,36 @@ public class ProcessProducerTest {
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 TOPIC_NAME = "FALCON.PROCESS";
+ private static final String ENTITY_NAME = "process1";
private BrokerService broker;
private volatile AssertionError error;
@BeforeClass
public void setup() throws Exception {
- args = new String[]{"-" + ARG.entityName.getArgName(), TOPIC_NAME,
- "-" + ARG.feedNames.getArgName(), "click-logs,raw-logs",
- "-" + ARG.feedInstancePaths.getArgName(),
- "/click-logs/10/05/05/00/20,/raw-logs/10/05/05/00/20",
- "-" + ARG.workflowId.getArgName(), "workflow-01-00",
- "-" + ARG.workflowUser.getArgName(), "falcon",
- "-" + ARG.runId.getArgName(), "1",
- "-" + ARG.nominalTime.getArgName(), "2011-01-01-01-00",
- "-" + ARG.timeStamp.getArgName(), "2012-01-01-01-00",
- "-" + ARG.brokerUrl.getArgName(), BROKER_URL,
- "-" + ARG.brokerImplClass.getArgName(), (BROKER_IMPL_CLASS),
- "-" + ARG.entityType.getArgName(), ("process"),
- "-" + ARG.operation.getArgName(), ("GENERATE"),
- "-" + ARG.logFile.getArgName(), ("/logFile"),
- "-" + ARG.topicName.getArgName(), (TOPIC_NAME),
- "-" + ARG.status.getArgName(), ("SUCCEEDED"),
- "-" + ARG.brokerTTL.getArgName(), "10",
- "-" + ARG.cluster.getArgName(), "corp", };
+ args = new String[] {
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), ENTITY_NAME,
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), "click-logs,raw-logs",
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(),
+ "/click-logs/10/05/05/00/20,/raw-logs/10/05/05/00/20",
+ "-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
+ "-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), "falcon",
+ "-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
+ "-" + WorkflowExecutionArgs.NOMINAL_TIME.getName(), "2011-01-01-01-00",
+ "-" + WorkflowExecutionArgs.TIMESTAMP.getName(), "2012-01-01-01-00",
+ "-" + WorkflowExecutionArgs.BRKR_URL.getName(), BROKER_URL,
+ "-" + WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), BROKER_IMPL_CLASS,
+ "-" + WorkflowExecutionArgs.USER_BRKR_URL.getName(), BROKER_URL,
+ "-" + WorkflowExecutionArgs.USER_BRKR_IMPL_CLASS.getName(), BROKER_IMPL_CLASS,
+ "-" + WorkflowExecutionArgs.ENTITY_TYPE.getName(), "process",
+ "-" + WorkflowExecutionArgs.OPERATION.getName(), "GENERATE",
+ "-" + WorkflowExecutionArgs.LOG_FILE.getName(), "/logFile",
+ "-" + WorkflowExecutionArgs.LOG_DIR.getName(), "/falcon/feed/agg-logs/",
+ "-" + WorkflowExecutionArgs.STATUS.getName(), "SUCCEEDED",
+ "-" + WorkflowExecutionArgs.BRKR_TTL.getName(), "10",
+ "-" + WorkflowExecutionArgs.CLUSTER_NAME.getName(), "corp",
+ };
+
broker = new BrokerService();
broker.addConnector(BROKER_URL);
broker.setDataDirectory("target/activemq");
@@ -92,7 +104,13 @@ public class ProcessProducerTest {
};
t.start();
Thread.sleep(100);
- new MessageProducer().run(this.args);
+
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(
+ args, WorkflowExecutionContext.Type.POST_PROCESSING);
+ JMSMessageProducer jmsMessageProducer = JMSMessageProducer.builder(context)
+ .type(JMSMessageProducer.MessageType.USER).build();
+ jmsMessageProducer.sendMessage();
+
t.join();
if (error != null) {
throw error;
@@ -100,26 +118,23 @@ public class ProcessProducerTest {
}
private void consumer() throws JMSException {
- ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
- BROKER_URL);
+ ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
Connection connection = connectionFactory.createConnection();
connection.start();
- Session session = connection.createSession(false,
- Session.AUTO_ACKNOWLEDGE);
- Destination destination = session.createTopic(TOPIC_NAME);
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createTopic(getTopicName());
MessageConsumer consumer = session.createConsumer(destination);
- // wait till you get atleast one message
+ // 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(ARG.feedNames.getArgName()),
- "click-logs");
- Assert.assertEquals(m.getString(ARG.feedInstancePaths.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_NAMES.getName()), "click-logs");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName()),
"/click-logs/10/05/05/00/20");
for (m = null; m == null;) {
@@ -127,24 +142,27 @@ public class ProcessProducerTest {
}
System.out.println("Consumed: " + m.toString());
assertMessage(m);
- Assert.assertEquals(m.getString(ARG.feedNames.getArgName()), "raw-logs");
- Assert.assertEquals(m.getString(ARG.feedInstancePaths.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_NAMES.getName()), "raw-logs");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName()),
"/raw-logs/10/05/05/00/20");
connection.close();
}
+ private String getTopicName() {
+ return JMSMessageProducer.FALCON_TOPIC_PREFIX + ENTITY_NAME;
+ }
+
private void assertMessage(MapMessage m) throws JMSException {
- Assert.assertEquals(m.getString(ARG.entityName.getArgName()),
- TOPIC_NAME);
- Assert.assertEquals(m.getString(ARG.workflowId.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.ENTITY_NAME.getName()), ENTITY_NAME);
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.WORKFLOW_ID.getName()),
"workflow-01-00");
- Assert.assertEquals(m.getString(ARG.workflowUser.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.WORKFLOW_USER.getName()),
"falcon");
- Assert.assertEquals(m.getString(ARG.runId.getArgName()), "1");
- Assert.assertEquals(m.getString(ARG.nominalTime.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.RUN_ID.getName()), "1");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.NOMINAL_TIME.getName()),
"2011-01-01T01:00Z");
- Assert.assertEquals(m.getString(ARG.timeStamp.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.TIMESTAMP.getName()),
"2012-01-01T01:00Z");
- Assert.assertEquals(m.getString(ARG.status.getArgName()), "SUCCEEDED");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.STATUS.getName()), "SUCCEEDED");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java b/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java
index 8744ad8..6e80db8 100644
--- a/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java
+++ b/metrics/src/main/java/org/apache/falcon/util/ResourcesReflectionUtil.java
@@ -43,7 +43,7 @@ public final class ResourcesReflectionUtil {
buildAnnotationsMapForClass("org.apache.falcon.resource.proxy.SchedulableEntityManagerProxy");
buildAnnotationsMapForClass("org.apache.falcon.resource.proxy.InstanceManagerProxy");
buildAnnotationsMapForClass("org.apache.falcon.resource.AbstractInstanceManager");
- buildAnnotationsMapForClass("org.apache.falcon.service.FalconTopicSubscriber");
+ buildAnnotationsMapForClass("org.apache.falcon.messaging.JMSMessageConsumer");
buildAnnotationsMapForClass("org.apache.falcon.aspect.GenericAlert");
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/oozie/src/main/java/org/apache/falcon/oozie/OozieCoordinatorBuilder.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/oozie/OozieCoordinatorBuilder.java b/oozie/src/main/java/org/apache/falcon/oozie/OozieCoordinatorBuilder.java
index 44f4410..459307c 100644
--- a/oozie/src/main/java/org/apache/falcon/oozie/OozieCoordinatorBuilder.java
+++ b/oozie/src/main/java/org/apache/falcon/oozie/OozieCoordinatorBuilder.java
@@ -26,7 +26,6 @@ import org.apache.falcon.entity.ExternalId;
import org.apache.falcon.entity.v0.Entity;
import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.feed.Feed;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
import org.apache.falcon.oozie.coordinator.CONFIGURATION;
import org.apache.falcon.oozie.coordinator.CONFIGURATION.Property;
import org.apache.falcon.oozie.coordinator.COORDINATORAPP;
@@ -36,6 +35,7 @@ import org.apache.falcon.oozie.feed.FeedRetentionCoordinatorBuilder;
import org.apache.falcon.oozie.process.ProcessExecutionCoordinatorBuilder;
import org.apache.falcon.util.OozieUtils;
import org.apache.falcon.util.StartupProperties;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
import org.apache.hadoop.fs.Path;
import org.apache.oozie.client.OozieClient;
@@ -62,7 +62,7 @@ public abstract class OozieCoordinatorBuilder<T extends Entity> extends OozieEnt
this.lifecycle = tag;
}
- public static final OozieCoordinatorBuilder get(Entity entity, Tag tag) {
+ public static OozieCoordinatorBuilder get(Entity entity, Tag tag) {
switch(entity.getEntityType()) {
case FEED:
switch (tag) {
@@ -94,28 +94,30 @@ public abstract class OozieCoordinatorBuilder<T extends Entity> extends OozieEnt
return EntityUtil.getWorkflowName(lifecycle, entity).toString();
}
- protected Path marshal(Cluster cluster, COORDINATORAPP coord, Path outPath) throws FalconException {
+ protected Path marshal(Cluster cluster, COORDINATORAPP coord,
+ Path outPath) throws FalconException {
return marshal(cluster, new ObjectFactory().createCoordinatorApp(coord),
OozieUtils.COORD_JAXB_CONTEXT, new Path(outPath, "coordinator.xml"));
}
- protected Properties createCoordDefaultConfiguration(Cluster cluster, String coordName) throws FalconException {
+ protected Properties createCoordDefaultConfiguration(Cluster cluster,
+ String coordName) throws FalconException {
Properties props = new Properties();
- props.put(ARG.entityName.getPropName(), entity.getName());
- props.put(ARG.nominalTime.getPropName(), NOMINAL_TIME_EL);
- props.put(ARG.timeStamp.getPropName(), ACTUAL_TIME_EL);
+ props.put(WorkflowExecutionArgs.ENTITY_NAME.getName(), entity.getName());
+ props.put(WorkflowExecutionArgs.NOMINAL_TIME.getName(), NOMINAL_TIME_EL);
+ props.put(WorkflowExecutionArgs.TIMESTAMP.getName(), ACTUAL_TIME_EL);
props.put("userBrokerUrl", ClusterHelper.getMessageBrokerUrl(cluster));
props.put("userBrokerImplClass", ClusterHelper.getMessageBrokerImplClass(cluster));
- String falconBrokerUrl = StartupProperties.get().getProperty(ARG.brokerUrl.getPropName(),
- "tcp://localhost:61616?daemon=true");
- props.put(ARG.brokerUrl.getPropName(), falconBrokerUrl);
- String falconBrokerImplClass = StartupProperties.get().getProperty(ARG.brokerImplClass.getPropName(),
- ClusterHelper.DEFAULT_BROKER_IMPL_CLASS);
- props.put(ARG.brokerImplClass.getPropName(), falconBrokerImplClass);
+ String falconBrokerUrl = StartupProperties.get().getProperty(
+ WorkflowExecutionArgs.BRKR_URL.getName(), "tcp://localhost:61616?daemon=true");
+ props.put(WorkflowExecutionArgs.BRKR_URL.getName(), falconBrokerUrl);
+ String falconBrokerImplClass = StartupProperties.get().getProperty(
+ WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), ClusterHelper.DEFAULT_BROKER_IMPL_CLASS);
+ props.put(WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), falconBrokerImplClass);
String jmsMessageTTL = StartupProperties.get().getProperty("broker.ttlInMins",
DEFAULT_BROKER_MSG_TTL.toString());
- props.put(ARG.brokerTTL.getPropName(), jmsMessageTTL);
- props.put(ARG.entityType.getPropName(), entity.getEntityType().name());
+ props.put(WorkflowExecutionArgs.BRKR_TTL.getName(), jmsMessageTTL);
+ props.put(WorkflowExecutionArgs.ENTITY_TYPE.getName(), entity.getEntityType().name());
props.put("logDir", getLogDirectory(cluster));
props.put(OozieClient.EXTERNAL_ID,
new ExternalId(entity.getName(), EntityUtil.getWorkflowNameTag(coordName, entity),
@@ -132,7 +134,7 @@ public abstract class OozieCoordinatorBuilder<T extends Entity> extends OozieEnt
props.put("entityName", entity.getName());
props.put("entityType", entity.getEntityType().name().toLowerCase());
- props.put(ARG.cluster.getPropName(), cluster.getName());
+ props.put(WorkflowExecutionArgs.CLUSTER_NAME.getName(), cluster.getName());
props.put(MR_QUEUE_NAME, "default");
props.put(MR_JOB_PRIORITY, "NORMAL");
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedReplicationCoordinatorBuilder.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedReplicationCoordinatorBuilder.java b/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedReplicationCoordinatorBuilder.java
index 2798db2..d6115b2 100644
--- a/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedReplicationCoordinatorBuilder.java
+++ b/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedReplicationCoordinatorBuilder.java
@@ -37,7 +37,6 @@ import org.apache.falcon.entity.v0.feed.Feed;
import org.apache.falcon.entity.v0.feed.LocationType;
import org.apache.falcon.expression.ExpressionHelper;
import org.apache.falcon.hadoop.HadoopClientFactory;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
import org.apache.falcon.oozie.OozieCoordinatorBuilder;
import org.apache.falcon.oozie.OozieOrchestrationWorkflowBuilder;
import org.apache.falcon.oozie.coordinator.COORDINATORAPP;
@@ -45,6 +44,7 @@ import org.apache.falcon.oozie.coordinator.SYNCDATASET;
import org.apache.falcon.oozie.coordinator.WORKFLOW;
import org.apache.falcon.oozie.coordinator.ACTION;
import org.apache.falcon.util.RuntimeProperties;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -249,7 +249,8 @@ public class FeedReplicationCoordinatorBuilder extends OozieCoordinatorBuilder<F
props.put("sourceRelativePaths", IGNORE); // this will bot be used for Table storage.
}
- private void propagateLateDataProperties(String instancePaths, String falconFeedStorageType, Properties props) {
+ private void propagateLateDataProperties(String instancePaths,
+ String falconFeedStorageType, Properties props) {
// todo these pairs are the same but used in different context
// late data handler - should-record action
props.put("falconInputFeeds", entity.getName());
@@ -260,11 +261,12 @@ public class FeedReplicationCoordinatorBuilder extends OozieCoordinatorBuilder<F
props.put("falconInputFeedStorageTypes", falconFeedStorageType);
// falcon post processing
- props.put(ARG.feedNames.getPropName(), entity.getName());
- props.put(ARG.feedInstancePaths.getPropName(), "${coord:dataOut('output')}");
+ props.put(WorkflowExecutionArgs.FEED_NAMES.getName(), entity.getName());
+ props.put(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(), "${coord:dataOut('output')}");
}
- private void setupHiveConfiguration(Cluster srcCluster, Cluster trgCluster, Path buildPath) throws FalconException {
+ private void setupHiveConfiguration(Cluster srcCluster, Cluster trgCluster,
+ Path buildPath) throws FalconException {
Configuration conf = ClusterHelper.getConfiguration(trgCluster);
FileSystem fs = HadoopClientFactory.get().createFileSystem(conf);
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedRetentionCoordinatorBuilder.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedRetentionCoordinatorBuilder.java b/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedRetentionCoordinatorBuilder.java
index 2238778..2a67fd3 100644
--- a/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedRetentionCoordinatorBuilder.java
+++ b/oozie/src/main/java/org/apache/falcon/oozie/feed/FeedRetentionCoordinatorBuilder.java
@@ -26,14 +26,14 @@ import org.apache.falcon.entity.v0.Frequency.TimeUnit;
import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.feed.Feed;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
-import org.apache.falcon.messaging.EntityInstanceMessage.EntityOps;
import org.apache.falcon.oozie.OozieCoordinatorBuilder;
import org.apache.falcon.oozie.OozieEntityBuilder;
import org.apache.falcon.oozie.OozieOrchestrationWorkflowBuilder;
import org.apache.falcon.oozie.coordinator.ACTION;
import org.apache.falcon.oozie.coordinator.COORDINATORAPP;
import org.apache.falcon.oozie.coordinator.WORKFLOW;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
import org.apache.hadoop.fs.Path;
import java.util.Arrays;
@@ -85,9 +85,10 @@ public class FeedRetentionCoordinatorBuilder extends OozieCoordinatorBuilder<Fee
props.put("limit", feedCluster.getRetention().getLimit().toString());
- props.put(ARG.operation.getPropName(), EntityOps.DELETE.name());
- props.put(ARG.feedNames.getPropName(), entity.getName());
- props.put(ARG.feedInstancePaths.getPropName(), IGNORE);
+ props.put(WorkflowExecutionArgs.OPERATION.getName(),
+ WorkflowExecutionContext.EntityOperations.DELETE.name());
+ props.put(WorkflowExecutionArgs.FEED_NAMES.getName(), entity.getName());
+ props.put(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(), IGNORE);
props.put("falconInputFeeds", entity.getName());
props.put("falconInPaths", IGNORE);
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/oozie/src/main/java/org/apache/falcon/oozie/process/ProcessExecutionCoordinatorBuilder.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/oozie/process/ProcessExecutionCoordinatorBuilder.java b/oozie/src/main/java/org/apache/falcon/oozie/process/ProcessExecutionCoordinatorBuilder.java
index e907087..d391032 100644
--- a/oozie/src/main/java/org/apache/falcon/oozie/process/ProcessExecutionCoordinatorBuilder.java
+++ b/oozie/src/main/java/org/apache/falcon/oozie/process/ProcessExecutionCoordinatorBuilder.java
@@ -37,7 +37,6 @@ import org.apache.falcon.entity.v0.process.Output;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.entity.v0.process.Workflow;
import org.apache.falcon.expression.ExpressionHelper;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
import org.apache.falcon.oozie.OozieCoordinatorBuilder;
import org.apache.falcon.oozie.OozieEntityBuilder;
import org.apache.falcon.oozie.OozieOrchestrationWorkflowBuilder;
@@ -50,6 +49,7 @@ import org.apache.falcon.oozie.coordinator.INPUTEVENTS;
import org.apache.falcon.oozie.coordinator.OUTPUTEVENTS;
import org.apache.falcon.oozie.coordinator.SYNCDATASET;
import org.apache.falcon.oozie.coordinator.WORKFLOW;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
import org.apache.hadoop.fs.Path;
import java.util.ArrayList;
@@ -243,8 +243,8 @@ public class ProcessExecutionCoordinatorBuilder extends OozieCoordinatorBuilder<
private void initializeOutputPaths(Cluster cluster, COORDINATORAPP coord, Properties props) throws FalconException {
if (entity.getOutputs() == null) {
- props.put(ARG.feedNames.getPropName(), "NONE");
- props.put(ARG.feedInstancePaths.getPropName(), IGNORE);
+ props.put(WorkflowExecutionArgs.FEED_NAMES.getName(), "NONE");
+ props.put(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(), IGNORE);
return;
}
@@ -282,8 +282,8 @@ public class ProcessExecutionCoordinatorBuilder extends OozieCoordinatorBuilder<
}
// Output feed name and path for parent workflow
- props.put(ARG.feedNames.getPropName(), StringUtils.join(outputFeeds, ','));
- props.put(ARG.feedInstancePaths.getPropName(), StringUtils.join(outputPaths, ','));
+ props.put(WorkflowExecutionArgs.FEED_NAMES.getName(), StringUtils.join(outputFeeds, ','));
+ props.put(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(), StringUtils.join(outputPaths, ','));
}
private DATAOUT createDataOut(Output output) {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/oozie/src/test/java/org/apache/falcon/oozie/feed/OozieFeedWorkflowBuilderTest.java
----------------------------------------------------------------------
diff --git a/oozie/src/test/java/org/apache/falcon/oozie/feed/OozieFeedWorkflowBuilderTest.java b/oozie/src/test/java/org/apache/falcon/oozie/feed/OozieFeedWorkflowBuilderTest.java
index fc1af7b..927aba3 100644
--- a/oozie/src/test/java/org/apache/falcon/oozie/feed/OozieFeedWorkflowBuilderTest.java
+++ b/oozie/src/test/java/org/apache/falcon/oozie/feed/OozieFeedWorkflowBuilderTest.java
@@ -216,7 +216,7 @@ public class OozieFeedWorkflowBuilderTest extends AbstractTestBase {
// verify workflow params
Assert.assertEquals(props.get("userWorkflowName"), "replication-policy");
- Assert.assertEquals(props.get("userWorkflowVersion"), "0.5");
+ Assert.assertEquals(props.get("userWorkflowVersion"), "0.6");
Assert.assertEquals(props.get("userWorkflowEngine"), "falcon");
// verify default params
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/oozie/src/test/java/org/apache/falcon/oozie/process/OozieProcessWorkflowBuilderTest.java
----------------------------------------------------------------------
diff --git a/oozie/src/test/java/org/apache/falcon/oozie/process/OozieProcessWorkflowBuilderTest.java b/oozie/src/test/java/org/apache/falcon/oozie/process/OozieProcessWorkflowBuilderTest.java
index ffcc88a..45a8732 100644
--- a/oozie/src/test/java/org/apache/falcon/oozie/process/OozieProcessWorkflowBuilderTest.java
+++ b/oozie/src/test/java/org/apache/falcon/oozie/process/OozieProcessWorkflowBuilderTest.java
@@ -40,7 +40,6 @@ import org.apache.falcon.entity.v0.process.Output;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.entity.v0.process.Validity;
import org.apache.falcon.entity.v0.process.Workflow;
-import org.apache.falcon.messaging.EntityInstanceMessage;
import org.apache.falcon.oozie.OozieEntityBuilder;
import org.apache.falcon.oozie.OozieOrchestrationWorkflowBuilder;
import org.apache.falcon.oozie.bundle.BUNDLEAPP;
@@ -55,6 +54,7 @@ import org.apache.falcon.security.CurrentUser;
import org.apache.falcon.security.SecurityUtil;
import org.apache.falcon.util.OozieUtils;
import org.apache.falcon.util.StartupProperties;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -691,13 +691,13 @@ public class OozieProcessWorkflowBuilderTest extends AbstractTestBase {
Assert.assertEquals(props.get("logDir"), getLogPath(processEntity));
String[] expected = {
- EntityInstanceMessage.ARG.feedNames.getPropName(),
- EntityInstanceMessage.ARG.feedInstancePaths.getPropName(),
- "falconInputFeeds",
+ WorkflowExecutionArgs.FEED_NAMES.getName(),
+ WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(),
+ WorkflowExecutionArgs.INPUT_FEED_NAMES.getName(),
"falconInPaths",
- "userWorkflowName",
- "userWorkflowVersion",
- "userWorkflowEngine",
+ WorkflowExecutionArgs.USER_WORKFLOW_NAME.getName(),
+ WorkflowExecutionArgs.USER_WORKFLOW_VERSION.getName(),
+ WorkflowExecutionArgs.USER_WORKFLOW_ENGINE.getName(),
};
for (String property : expected) {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/oozie/src/test/resources/oozie/xmls/workflow.xml
----------------------------------------------------------------------
diff --git a/oozie/src/test/resources/oozie/xmls/workflow.xml b/oozie/src/test/resources/oozie/xmls/workflow.xml
index 5d14413..81807a6 100644
--- a/oozie/src/test/resources/oozie/xmls/workflow.xml
+++ b/oozie/src/test/resources/oozie/xmls/workflow.xml
@@ -22,7 +22,7 @@
<java>
<job-tracker>${jobTracker}</job-tracker>
<name-node>${nameNode}</name-node>
- <main-class>org.apache.falcon.messaging.MessageProducer</main-class>
+ <main-class>org.apache.falcon.messaging.JMSMessageProducer</main-class>
<arg>${wf:name()}</arg>
<arg>${wf:appPath()}</arg>
<arg>${timestamp()}</arg>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/prism/src/main/java/org/apache/falcon/service/FalconTopicSubscriber.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/service/FalconTopicSubscriber.java b/prism/src/main/java/org/apache/falcon/service/FalconTopicSubscriber.java
deleted file mode 100644
index 35e0ba3..0000000
--- a/prism/src/main/java/org/apache/falcon/service/FalconTopicSubscriber.java
+++ /dev/null
@@ -1,200 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-package org.apache.falcon.service;
-
-import org.apache.falcon.FalconException;
-import org.apache.falcon.aspect.GenericAlert;
-import org.apache.falcon.entity.EntityUtil;
-import org.apache.falcon.entity.v0.Entity;
-import org.apache.falcon.entity.v0.SchemaHelper;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
-import org.apache.falcon.metadata.MetadataMappingService;
-import org.apache.falcon.rerun.event.RerunEvent.RerunType;
-import org.apache.falcon.rerun.handler.AbstractRerunHandler;
-import org.apache.falcon.rerun.handler.RerunHandlerFactory;
-import org.apache.falcon.resource.InstancesResult;
-import org.apache.falcon.security.CurrentUser;
-import org.apache.falcon.workflow.WorkflowEngineFactory;
-import org.apache.falcon.workflow.engine.AbstractWorkflowEngine;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.jms.*;
-import java.lang.reflect.InvocationTargetException;
-import java.util.Date;
-
-/**
- * Subscribes to the falcon topic for handling retries and alerts.
- */
-public class FalconTopicSubscriber implements MessageListener, ExceptionListener {
- private static final Logger LOG = LoggerFactory.getLogger(FalconTopicSubscriber.class);
-
- private final String implementation;
- private final String userName;
- private final String password;
- private final String url;
- private final String topicName;
-
- private Connection connection;
- private TopicSubscriber subscriber;
-
- private AbstractRerunHandler retryHandler = RerunHandlerFactory.getRerunHandler(RerunType.RETRY);
- private AbstractRerunHandler latedataHandler = RerunHandlerFactory.getRerunHandler(RerunType.LATE);
-
- public FalconTopicSubscriber(String implementation, String userName,
- String password, String url, String topicName) {
- this.implementation = implementation;
- this.userName = userName;
- this.password = password;
- this.url = url;
- this.topicName = topicName;
- }
-
- public void startSubscriber() throws FalconException {
- try {
- connection = createAndGetConnection(implementation, userName, password, url);
- TopicSession session = (TopicSession) connection.createSession(
- false, Session.AUTO_ACKNOWLEDGE);
- Topic destination = session.createTopic(topicName);
- subscriber = session.createSubscriber(destination);
- subscriber.setMessageListener(this);
- connection.setExceptionListener(this);
- connection.start();
- } catch (Exception e) {
- LOG.error("Error starting subscriber of topic: {}", this, e);
- throw new FalconException(e);
- }
- }
-
- @Override
- public void onMessage(Message message) {
- MapMessage mapMessage = (MapMessage) message;
- try {
- if (LOG.isDebugEnabled()) {debug(mapMessage); }
- String cluster = mapMessage.getString(ARG.cluster.getArgName());
- String entityName = mapMessage.getString(ARG.entityName.getArgName());
- String entityType = mapMessage.getString(ARG.entityType.getArgName());
- String workflowId = mapMessage.getString(ARG.workflowId.getArgName());
- String workflowUser = mapMessage.getString(ARG.workflowUser.getArgName());
- String runId = mapMessage.getString(ARG.runId.getArgName());
- String nominalTime = mapMessage.getString(ARG.nominalTime.getArgName());
- String status = mapMessage.getString(ARG.status.getArgName());
- String operation = mapMessage.getString(ARG.operation.getArgName());
-
- CurrentUser.authenticate(workflowUser);
- AbstractWorkflowEngine wfEngine = WorkflowEngineFactory.getWorkflowEngine();
- InstancesResult result = wfEngine.getJobDetails(cluster, workflowId);
- Date startTime = result.getInstances()[0].startTime;
- Date endTime = result.getInstances()[0].endTime;
- Long duration = (endTime.getTime() - startTime.getTime()) * 1000000;
-
- if (status.equalsIgnoreCase("FAILED")) {
- retryHandler.handleRerun(cluster, entityType, entityName,
- nominalTime, runId, workflowId, workflowUser,
- System.currentTimeMillis());
-
- GenericAlert.instrumentFailedInstance(cluster, entityType,
- entityName, nominalTime, workflowId, workflowUser, runId, operation,
- SchemaHelper.formatDateUTC(startTime), "", "", duration);
-
- } else if (status.equalsIgnoreCase("SUCCEEDED")) {
- Entity entity = EntityUtil.getEntity(entityType, entityName);
- //late data handling not applicable for feed retention action
- if (!operation.equalsIgnoreCase("DELETE") && EntityUtil.getLateProcess(entity) != null) {
- latedataHandler.handleRerun(cluster, entityType, entityName,
- nominalTime, runId, workflowId, workflowUser,
- System.currentTimeMillis());
- } else {
- LOG.info("Late data handling not applicable for entityType: {}, entityName: {} operation: {}",
- entityType, entityName, operation);
- }
- GenericAlert.instrumentSucceededInstance(cluster, entityType,
- entityName, nominalTime, workflowId, workflowUser, runId, operation,
- SchemaHelper.formatDateUTC(startTime), duration);
-
- notifyMetadataMappingService(entityName, operation, mapMessage.getString(ARG.logDir.getArgName()));
- }
- } catch (JMSException e) {
- LOG.info("Error in onMessage for subscriber of topic: {}", this, e);
- } catch (FalconException e) {
- LOG.info("Error in onMessage for subscriber of topic: {}", this, e);
- } catch (Exception e) {
- LOG.info("Error in onMessage for subscriber of topic: {}", this, e);
- }
- }
-
- private void notifyMetadataMappingService(String entityName, String operation,
- String logDir) throws FalconException {
- if (Services.get().isRegistered(MetadataMappingService.SERVICE_NAME)) {
- MetadataMappingService service = Services.get().getService(MetadataMappingService.SERVICE_NAME);
- service.onSuccessfulWorkflowCompletion(entityName, operation, logDir);
- }
- }
-
- private void debug(MapMessage mapMessage) throws JMSException {
- StringBuilder buff = new StringBuilder();
- buff.append("Received:{");
- for (ARG arg : ARG.values()) {
- buff.append(arg.getArgName()).append('=')
- .append(mapMessage.getString(arg.getArgName())).append(", ");
- }
- buff.append("}");
- LOG.debug(buff.toString());
- }
-
- @Override
- public void onException(JMSException ignore) {
- LOG.info("Error in onException for subscriber of topic: {}", this.toString(), ignore);
- }
-
- public void closeSubscriber() throws FalconException {
- try {
- LOG.info("Closing subscriber on topic: {}", this.topicName);
- if (subscriber != null) {
- subscriber.close();
- }
- if (connection != null) {
- connection.close();
- }
- } catch (JMSException e) {
- LOG.error("Error closing subscriber of topic: {}", this.toString(), e);
- throw new FalconException(e);
- }
- }
-
- private static Connection createAndGetConnection(String implementation,
- String userName, String password, String url)
- throws JMSException, ClassNotFoundException, InstantiationException,
- IllegalAccessException, InvocationTargetException, NoSuchMethodException {
-
- @SuppressWarnings("unchecked")
- Class<ConnectionFactory> clazz = (Class<ConnectionFactory>) FalconTopicSubscriber.class
- .getClassLoader().loadClass(implementation);
-
- ConnectionFactory connectionFactory = clazz.getConstructor(
- String.class, String.class, String.class).newInstance(userName,
- password, url);
-
- return connectionFactory.createConnection();
- }
-
- @Override
- public String toString() {
- return topicName;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java b/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java
index 9ae1ad1..06fec9f 100644
--- a/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java
+++ b/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java
@@ -18,14 +18,16 @@
package org.apache.falcon.service;
import org.apache.falcon.FalconException;
+import org.apache.falcon.messaging.JMSMessageConsumer;
import org.apache.falcon.util.StartupProperties;
+import org.apache.falcon.workflow.WorkflowJobEndNotificationService;
/**
* A Falcon Service that initializes and starts a topic subscriber.
*/
public class ProcessSubscriberService implements FalconService {
- private FalconTopicSubscriber subscriber;
+ private JMSMessageConsumer subscriber;
private static enum JMSProps {
FalconBrokerImplClass("broker.impl.class", "org.apache.activemq.ActiveMQConnectionFactory"),
@@ -48,11 +50,18 @@ public class ProcessSubscriberService implements FalconService {
@Override
public void init() throws FalconException {
+ if (!Services.get().isRegistered(WorkflowJobEndNotificationService.NAME)) {
+ throw new FalconException("WorkflowJobEndNotificationService must be configured ahead");
+ }
+
String falconBrokerImplClass = getPropertyValue(JMSProps.FalconBrokerImplClass);
String falconBrokerUrl = getPropertyValue(JMSProps.FalconBrokerUrl);
String falconEntityTopic = getPropertyValue(JMSProps.FalconEntityTopic);
- subscriber = new FalconTopicSubscriber(falconBrokerImplClass, "", "", falconBrokerUrl, falconEntityTopic);
+ WorkflowJobEndNotificationService jobEndNotificationService =
+ Services.get().getService(WorkflowJobEndNotificationService.NAME);
+ subscriber = new JMSMessageConsumer(falconBrokerImplClass, "", "", falconBrokerUrl,
+ falconEntityTopic, jobEndNotificationService);
subscriber.startSubscriber();
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/prism/src/test/java/org/apache/falcon/service/FalconTopicSubscriberTest.java
----------------------------------------------------------------------
diff --git a/prism/src/test/java/org/apache/falcon/service/FalconTopicSubscriberTest.java b/prism/src/test/java/org/apache/falcon/service/FalconTopicSubscriberTest.java
deleted file mode 100644
index 9b1d42a..0000000
--- a/prism/src/test/java/org/apache/falcon/service/FalconTopicSubscriberTest.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-package org.apache.falcon.service;
-
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.falcon.messaging.EntityInstanceMessage;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
-import org.mortbay.log.Log;
-import org.testng.Assert;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.Test;
-
-import javax.jms.*;
-
-/**
- * Test for FalconTopicSubscriber.
- */
-public class FalconTopicSubscriberTest {
-
- private static final String BROKER_URL = "vm://localhost";
- private static final String BROKER_IMPL_CLASS = "org.apache.activemq.ActiveMQConnectionFactory";
- private static final String TOPIC_NAME = "FALCON.ENTITY.TOPIC";
- private static final String SECONDARY_TOPIC_NAME = "FALCON.ENTITY.SEC.TOPIC";
- private BrokerService broker;
-
- @BeforeClass
- public void setup() throws Exception {
- broker = new BrokerService();
- broker.addConnector(BROKER_URL);
- broker.setDataDirectory("target/activemq");
- broker.setBrokerName("localhost");
- broker.start();
- }
-
- public void sendMessages(String topic) throws JMSException {
- ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
- BROKER_URL);
- Connection connection = connectionFactory.createConnection();
- connection.start();
-
- Session session = connection.createSession(false,
- Session.AUTO_ACKNOWLEDGE);
- Topic destination = session.createTopic(topic);
- MessageProducer producer = session
- .createProducer(destination);
- producer.setDeliveryMode(DeliveryMode.PERSISTENT);
- for (int i = 0; i < 3; i++) {
- EntityInstanceMessage falconMessage = getMockFalconMessage(i);
- MapMessage message = session.createMapMessage();
- for (ARG arg : ARG.values()) {
- message.setString(arg.getPropName(), falconMessage
- .getKeyValueMap().get(arg));
- }
- Log.debug("Sending:" + message);
- producer.send(message);
- }
-
- EntityInstanceMessage message = getMockFalconMessage(15);
- MapMessage mapMessage = session.createMapMessage();
- message.getKeyValueMap().put(ARG.status, "FAILED");
- for (ARG arg : ARG.values()) {
- mapMessage.setString(arg.getPropName(), message.getKeyValueMap().get(arg));
- }
- producer.send(mapMessage);
- }
-
- private EntityInstanceMessage getMockFalconMessage(int i) {
- EntityInstanceMessage message = new EntityInstanceMessage();
- message.getKeyValueMap().put(ARG.brokerImplClass, BROKER_IMPL_CLASS);
- message.getKeyValueMap().put(ARG.brokerUrl, BROKER_URL);
- message.getKeyValueMap().put(ARG.entityName, "process1");
- message.getKeyValueMap().put(ARG.entityType, "PROCESS");
- message.getKeyValueMap().put(ARG.feedInstancePaths,
- "/clicks/hour/00/0" + i);
- message.getKeyValueMap().put(ARG.feedNames, "clicks");
- message.getKeyValueMap().put(ARG.logFile, "/logfile");
- message.getKeyValueMap().put(ARG.nominalTime, "2012-10-10-10-10");
- message.getKeyValueMap().put(ARG.operation, "GENERATE");
- message.getKeyValueMap().put(ARG.runId, "0");
- message.getKeyValueMap().put(ARG.timeStamp, "2012-10-10-10-1" + i);
- message.getKeyValueMap().put(ARG.workflowId, "workflow-" + i);
- message.getKeyValueMap().put(ARG.topicName, TOPIC_NAME);
- message.getKeyValueMap().put(ARG.status, "SUCCEEDED");
- message.getKeyValueMap().put(ARG.workflowUser, "falcon");
- return message;
- }
-
- @Test
- public void testSubscriber() throws Exception{
- //Comma separated topics are supported in startup properties
- FalconTopicSubscriber subscriber = new FalconTopicSubscriber(
- BROKER_IMPL_CLASS, "", "", BROKER_URL, TOPIC_NAME+","+SECONDARY_TOPIC_NAME);
- subscriber.startSubscriber();
- sendMessages(TOPIC_NAME);
- Assert.assertEquals(broker.getAdminView().getTotalEnqueueCount(), 9);
- sendMessages(SECONDARY_TOPIC_NAME);
- Assert.assertEquals(broker.getAdminView().getTotalEnqueueCount(), 17);
- Assert.assertEquals(broker.getAdminView().getTotalConsumerCount(), 2);
- subscriber.closeSubscriber();
- }
-
- @AfterClass
- public void tearDown() throws Exception {
- broker.deleteAllMessages();
- broker.stop();
- }
-}
[3/8] git commit: FALCON-485 - Simplify JMS Message Sender/Consumer
and use Workflow Context. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-485 - Simplify JMS Message Sender/Consumer and use Workflow Context. 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/23762e55
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/23762e55
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/23762e55
Branch: refs/heads/master
Commit: 23762e55b860fa1ecf2433de0cda654e2a3f0fc1
Parents: b42c53b
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Fri Aug 8 12:57:03 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Fri Aug 8 12:57:03 2014 -0700
----------------------------------------------------------------------
.../org/apache/falcon/entity/FeedHelper.java | 2 +-
.../workflow/WorkflowExecutionContext.java | 40 +-
.../falcon/messaging/EntityInstanceMessage.java | 242 ------------
.../messaging/EntityInstanceMessageCreator.java | 56 ---
.../falcon/messaging/JMSMessageConsumer.java | 181 +++++++++
.../falcon/messaging/JMSMessageProducer.java | 376 +++++++++++++++++++
.../falcon/messaging/MessageProducer.java | 180 ---------
.../messaging/FalconTopicProducerTest.java | 215 -----------
.../falcon/messaging/FeedProducerTest.java | 88 +++--
.../messaging/JMSMessageConsumerTest.java | 156 ++++++++
.../messaging/JMSMessageProducerTest.java | 218 +++++++++++
.../falcon/messaging/ProcessProducerTest.java | 100 +++--
.../falcon/util/ResourcesReflectionUtil.java | 2 +-
.../falcon/oozie/OozieCoordinatorBuilder.java | 34 +-
.../feed/FeedReplicationCoordinatorBuilder.java | 12 +-
.../feed/FeedRetentionCoordinatorBuilder.java | 11 +-
.../ProcessExecutionCoordinatorBuilder.java | 10 +-
.../feed/OozieFeedWorkflowBuilderTest.java | 2 +-
.../OozieProcessWorkflowBuilderTest.java | 14 +-
.../src/test/resources/oozie/xmls/workflow.xml | 2 +-
.../falcon/service/FalconTopicSubscriber.java | 200 ----------
.../service/ProcessSubscriberService.java | 13 +-
.../service/FalconTopicSubscriberTest.java | 124 ------
23 files changed, 1131 insertions(+), 1147 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/FeedHelper.java b/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
index 8c61ac2..d09a12f 100644
--- a/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
+++ b/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
@@ -284,7 +284,7 @@ public final class FeedHelper {
try {
version = BuildProperties.get().getProperty("build.version");
} catch (Exception e) { // unfortunate that this is only available in prism/webapp
- version = "0.5";
+ version = "0.6";
}
props.put("userWorkflowVersion", version);
return props;
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/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 60e6c58..637cc3e 100644
--- a/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionContext.java
+++ b/common/src/main/java/org/apache/falcon/workflow/WorkflowExecutionContext.java
@@ -73,6 +73,24 @@ public class WorkflowExecutionContext {
GENERATE, DELETE, ARCHIVE, REPLICATE, CHMOD
}
+ public static final WorkflowExecutionArgs[] USER_MESSAGE_ARGS = {
+ WorkflowExecutionArgs.CLUSTER_NAME,
+ WorkflowExecutionArgs.ENTITY_NAME,
+ WorkflowExecutionArgs.ENTITY_TYPE,
+ WorkflowExecutionArgs.NOMINAL_TIME,
+ WorkflowExecutionArgs.OPERATION,
+
+ WorkflowExecutionArgs.FEED_NAMES,
+ WorkflowExecutionArgs.FEED_INSTANCE_PATHS,
+
+ WorkflowExecutionArgs.WORKFLOW_ID,
+ WorkflowExecutionArgs.WORKFLOW_USER,
+ WorkflowExecutionArgs.RUN_ID,
+ WorkflowExecutionArgs.STATUS,
+ WorkflowExecutionArgs.TIMESTAMP,
+ WorkflowExecutionArgs.LOG_FILE,
+ };
+
private final Map<WorkflowExecutionArgs, String> context;
private final long creationTime;
@@ -287,10 +305,9 @@ public class WorkflowExecutionContext {
}
public static WorkflowExecutionContext create(String[] args, Type type) throws FalconException {
- try {
- Map<WorkflowExecutionArgs, String> wfProperties = new HashMap<WorkflowExecutionArgs, String>();
- wfProperties.put(WorkflowExecutionArgs.CONTEXT_TYPE, type.name());
+ Map<WorkflowExecutionArgs, String> wfProperties = new HashMap<WorkflowExecutionArgs, String>();
+ try {
CommandLine cmd = getCommand(args);
for (WorkflowExecutionArgs arg : WorkflowExecutionArgs.values()) {
String optionValue = arg.getOptionValue(cmd);
@@ -298,14 +315,16 @@ public class WorkflowExecutionContext {
wfProperties.put(arg, optionValue);
}
}
-
- wfProperties.put(WorkflowExecutionArgs.CONTEXT_FILE,
- getFilePath(wfProperties.get(WorkflowExecutionArgs.LOG_DIR),
- wfProperties.get(WorkflowExecutionArgs.ENTITY_NAME)));
- return new WorkflowExecutionContext(wfProperties);
} catch (ParseException e) {
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)));
+
+ return new WorkflowExecutionContext(wfProperties);
}
private static CommandLine getCommand(String[] arguments) throws ParseException {
@@ -323,4 +342,9 @@ public class WorkflowExecutionContext {
option.setRequired(isRequired);
options.addOption(option);
}
+
+ public static WorkflowExecutionContext create(Map<WorkflowExecutionArgs, String> wfProperties) {
+ wfProperties.put(WorkflowExecutionArgs.CONTEXT_TYPE, Type.POST_PROCESSING.name());
+ return new WorkflowExecutionContext(wfProperties);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessage.java
----------------------------------------------------------------------
diff --git a/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessage.java b/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessage.java
deleted file mode 100644
index 679e9ea..0000000
--- a/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessage.java
+++ /dev/null
@@ -1,242 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-
-package org.apache.falcon.messaging;
-
-import org.apache.commons.cli.CommandLine;
-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.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.text.DateFormat;
-import java.text.ParseException;
-import java.text.SimpleDateFormat;
-import java.util.Date;
-import java.util.LinkedHashMap;
-import java.util.Map;
-
-/**
- * Value Object which is stored in JMS Topic as MapMessage.
- */
-public class EntityInstanceMessage {
-
- private final Map<ARG, String> keyValueMap = new LinkedHashMap<ARG, String>();
- private static final Logger LOG = LoggerFactory.getLogger(EntityInstanceMessage.class);
- private static final String FALCON_ENTITY_TOPIC_NAME = "FALCON.ENTITY.TOPIC";
-
- /**
- * Feed Entity operations supported.
- */
- public enum EntityOps {
- GENERATE, DELETE, ARCHIVE, REPLICATE, CHMOD
- }
-
- /**
- * properties available in feed entity operation workflow.
- */
- public enum ARG {
- entityName("entityName"),
- feedNames("feedNames"),
- feedInstancePaths("feedInstancePaths"),
- workflowId("workflowId"),
- runId("runId"),
- nominalTime("nominalTime"),
- timeStamp("timeStamp"),
- brokerUrl("broker.url"),
- brokerImplClass("broker.impl.class"),
- entityType("entityType"),
- operation("operation"),
- logFile("logFile"),
- topicName("topicName"),
- status("status"),
- brokerTTL("broker.ttlInMins"),
- cluster("cluster"),
- workflowUser("workflowUser"),
- logDir("logDir");
-
- private String propName;
-
- private ARG(String propName) {
- this.propName = propName;
- }
-
- /**
- * @return Name of the Argument used in the parent workflow to pass
- * arguments to MessageProducer Main class.
- */
- public String getArgName() {
- return this.name();
- }
-
- /**
- * @return Name of the property used in the startup.properties,
- * coordinator and parent workflow.
- */
- public String getPropName() {
- return this.propName;
- }
- }
-
- public Map<ARG, String> getKeyValueMap() {
- return this.keyValueMap;
- }
-
- public String getTopicName() {
- return this.keyValueMap.get(ARG.topicName);
- }
-
- public String getFeedName() {
- return this.keyValueMap.get(ARG.feedNames);
- }
-
- public void setFeedName(String feedName) {
- this.keyValueMap.remove(ARG.feedNames);
- this.keyValueMap.put(ARG.feedNames, feedName);
- }
-
- public String getFeedInstancePath() {
- return this.keyValueMap.get(ARG.feedInstancePaths);
- }
-
- public void setFeedInstancePath(String feedInstancePath) {
- this.keyValueMap.remove(ARG.feedInstancePaths);
- this.keyValueMap.put(ARG.feedInstancePaths, feedInstancePath);
- }
-
- public String getEntityType() {
- return this.keyValueMap.get(ARG.entityType);
- }
-
- public String getBrokerTTL() {
- return this.keyValueMap.get(ARG.brokerTTL);
- }
-
- public void convertDateFormat() throws ParseException {
- String date = this.keyValueMap.remove(ARG.nominalTime);
- this.keyValueMap.put(ARG.nominalTime, getFalconDate(date));
- date = this.keyValueMap.remove(ARG.timeStamp);
- this.keyValueMap.put(ARG.timeStamp, getFalconDate(date));
- }
-
- public static EntityInstanceMessage[] getMessages(CommandLine cmd)
- throws ParseException {
-
- String[] feedNames = getFeedNames(cmd);
- if (feedNames == null) {
- return null;
- }
-
- String[] feedPaths;
- try {
- feedPaths = getFeedPaths(cmd);
- } catch (IOException e) {
- LOG.error("Error getting instance paths", e);
- throw new RuntimeException(e);
- }
-
- EntityInstanceMessage[] messages = new EntityInstanceMessage[feedPaths.length];
- for (int i = 0; i < feedPaths.length; i++) {
- EntityInstanceMessage message = new EntityInstanceMessage();
- setDefaultValues(cmd, message);
- // override default values
- if (message.getEntityType().equalsIgnoreCase("PROCESS")) {
- message.setFeedName(feedNames[i]);
- } else {
- message.setFeedName(message.getFeedName());
- }
- message.setFeedInstancePath(feedPaths[i]);
- message.convertDateFormat();
- messages[i] = message;
- }
-
- return messages;
- }
-
- private static void setDefaultValues(CommandLine cmd,
- EntityInstanceMessage message) {
- for (ARG arg : ARG.values()) {
- message.keyValueMap.put(arg, cmd.getOptionValue(arg.name()));
- }
- }
-
- private static String[] getFeedNames(CommandLine cmd) {
- String feedNameStr = cmd.getOptionValue(ARG.feedNames.getArgName());
- String topicName = cmd.getOptionValue(ARG.topicName.getArgName());
- if (topicName.equals(FALCON_ENTITY_TOPIC_NAME)) {
- return new String[]{feedNameStr};
- }
- if (feedNameStr.equals("null")) {
- return null;
- }
-
- return feedNameStr.split(",");
- }
-
- private static String[] getFeedPaths(CommandLine cmd) throws IOException {
- String topicName = cmd.getOptionValue(ARG.topicName.getArgName());
- String operation = cmd.getOptionValue(ARG.operation.getArgName());
-
- if (topicName.equals(FALCON_ENTITY_TOPIC_NAME)) {
- LOG.debug("Returning instance paths for Falcon Topic: {}",
- cmd.getOptionValue(ARG.feedInstancePaths.getArgName()));
- return new String[]{cmd.getOptionValue(ARG.feedInstancePaths.getArgName()), };
- }
-
- if (operation.equals(EntityOps.GENERATE.name()) || operation.equals(EntityOps.REPLICATE.name())) {
- LOG.debug("Returning instance paths: {}", cmd.getOptionValue(ARG.feedInstancePaths.getArgName()));
- return cmd.getOptionValue(ARG.feedInstancePaths.getArgName()).split(",");
- }
- //else case of feed retention
- Path logFile = new Path(cmd.getOptionValue(ARG.logFile.getArgName()));
- FileSystem fs = FileSystem.get(logFile.toUri(), new Configuration());
-
- if (!fs.exists(logFile)) {
- //Evictor Failed without deleting a single path
- return new String[0];
- }
-
- ByteArrayOutputStream writer = new ByteArrayOutputStream();
- InputStream instance = fs.open(logFile);
- IOUtils.copyBytes(instance, writer, 4096, true);
- String[] instancePaths = writer.toString().split("=");
- fs.delete(logFile, true);
- LOG.info("Deleted feed instance paths file: {}", logFile);
- if (instancePaths.length == 1) {
- LOG.debug("Returning 0 instance paths for feed ");
- return new String[0];
- } else {
- LOG.debug("Returning instance paths for feed {}", instancePaths[1]);
- return instancePaths[1].split(",");
- }
- }
-
- public String getFalconDate(String nominalTime) throws ParseException {
- DateFormat nominalFormat = new SimpleDateFormat(
- "yyyy'-'MM'-'dd'-'HH'-'mm");
- Date nominalDate = nominalFormat.parse(nominalTime);
- DateFormat falconFormat = new SimpleDateFormat(
- "yyyy'-'MM'-'dd'T'HH':'mm'Z'");
- return falconFormat.format(nominalDate);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessageCreator.java
----------------------------------------------------------------------
diff --git a/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessageCreator.java b/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessageCreator.java
deleted file mode 100644
index c8ea12d..0000000
--- a/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessageCreator.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-
-package org.apache.falcon.messaging;
-
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
-
-import javax.jms.JMSException;
-import javax.jms.MapMessage;
-import javax.jms.Message;
-import javax.jms.Session;
-import java.util.Map.Entry;
-
-/**
- * Falcon JMS message creator- creates JMS TextMessage.
- */
-public class EntityInstanceMessageCreator {
-
- private MapMessage mapMessage;
-
- private final EntityInstanceMessage instanceMessage;
-
- public EntityInstanceMessageCreator(EntityInstanceMessage instanceMessage) {
- this.instanceMessage = instanceMessage;
- }
-
- public Message createMessage(Session session) throws JMSException {
- mapMessage = session.createMapMessage();
- for (Entry<ARG, String> entry : instanceMessage.getKeyValueMap().entrySet()) {
- mapMessage.setString(entry.getKey().getArgName(),
- instanceMessage.getKeyValueMap().get(entry.getKey()));
- }
-
- return mapMessage;
- }
-
- @Override
- public String toString() {
- return this.mapMessage.toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageConsumer.java
----------------------------------------------------------------------
diff --git a/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageConsumer.java b/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageConsumer.java
new file mode 100644
index 0000000..573f5bd
--- /dev/null
+++ b/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageConsumer.java
@@ -0,0 +1,181 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.messaging;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.falcon.FalconException;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
+import org.apache.falcon.workflow.WorkflowJobEndNotificationService;
+import org.apache.log4j.Logger;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.Session;
+import javax.jms.Topic;
+import javax.jms.TopicSession;
+import javax.jms.TopicSubscriber;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Subscribes to the falcon topic for handling retries and alerts.
+ */
+public class JMSMessageConsumer implements MessageListener, ExceptionListener {
+ private static final Logger LOG = Logger.getLogger(JMSMessageConsumer.class);
+
+ private final String implementation;
+ private final String userName;
+ private final String password;
+ private final String url;
+ private final String topicName;
+
+ private Connection connection;
+ private TopicSubscriber subscriber;
+
+ private final WorkflowJobEndNotificationService jobEndNotificationService;
+
+ public JMSMessageConsumer(String implementation, String userName,
+ String password, String url, String topicName,
+ WorkflowJobEndNotificationService jobEndNotificationService) {
+ this.implementation = implementation;
+ this.userName = userName;
+ this.password = password;
+ this.url = url;
+ this.topicName = topicName;
+ this.jobEndNotificationService = jobEndNotificationService;
+ }
+
+ public void startSubscriber() throws FalconException {
+ try {
+ connection = createAndGetConnection(implementation, userName, password, url);
+ TopicSession session = (TopicSession) connection.createSession(
+ false, Session.AUTO_ACKNOWLEDGE);
+ Topic destination = session.createTopic(topicName);
+ subscriber = session.createSubscriber(destination);
+ subscriber.setMessageListener(this);
+ connection.setExceptionListener(this);
+ connection.start();
+ } catch (Exception e) {
+ LOG.error("Error starting subscriber of topic: " + this.toString(), e);
+ throw new FalconException(e);
+ }
+ }
+
+ @Override
+ public void onMessage(Message message) {
+ MapMessage mapMessage = (MapMessage) message;
+
+ try {
+ if (LOG.isDebugEnabled()) {debug(mapMessage); }
+
+ WorkflowExecutionContext context = createContext(mapMessage);
+ if (context.hasWorkflowFailed()) {
+ onFailure(context);
+ } else if (context.hasWorkflowSucceeded()) {
+ onSuccess(context);
+ }
+ } catch (JMSException e) {
+ LOG.info("Error in onMessage for subscriber of topic: " + this.toString(), e);
+ } catch (FalconException e) {
+ LOG.info("Error in onMessage for subscriber of topic: " + this.toString(), e);
+ } catch (Exception e) {
+ LOG.info("Error in onMessage for subscriber of topic: " + this.toString(), e);
+ }
+ }
+
+ private WorkflowExecutionContext createContext(MapMessage mapMessage) throws JMSException {
+ // for backwards compatibility, read all args from message
+ Map<WorkflowExecutionArgs, String> wfProperties = new HashMap<WorkflowExecutionArgs, String>();
+ for (WorkflowExecutionArgs arg : WorkflowExecutionArgs.values()) {
+ String optionValue = mapMessage.getString(arg.getName());
+ if (StringUtils.isNotEmpty(optionValue)) {
+ wfProperties.put(arg, optionValue);
+ }
+ }
+
+ return WorkflowExecutionContext.create(wfProperties);
+ }
+
+ public void onFailure(WorkflowExecutionContext context) throws FalconException {
+ jobEndNotificationService.notifyFailure(context);
+ }
+
+ public void onSuccess(WorkflowExecutionContext context) throws FalconException {
+ jobEndNotificationService.notifySuccess(context);
+ }
+
+ private void debug(MapMessage mapMessage) throws JMSException {
+ StringBuilder buff = new StringBuilder();
+ buff.append("Received:{");
+ for (WorkflowExecutionArgs arg : WorkflowExecutionArgs.values()) {
+ buff.append(arg.getName()).append('=')
+ .append(mapMessage.getString(arg.getName())).append(", ");
+ }
+ buff.append("}");
+ LOG.debug(buff);
+ }
+
+ @Override
+ public void onException(JMSException ignore) {
+ LOG.info("Error in onException for subscriber of topic: " + this.toString(), ignore);
+ }
+
+ public void closeSubscriber() throws FalconException {
+ try {
+ LOG.info("Closing subscriber on topic : " + this.topicName);
+ if (subscriber != null) {
+ subscriber.close();
+ }
+ if (connection != null) {
+ connection.close();
+ }
+ } catch (JMSException e) {
+ LOG.error("Error closing subscriber of topic: " + this.toString(), e);
+ throw new FalconException(e);
+ }
+ }
+
+ private static Connection createAndGetConnection(String implementation,
+ String userName, String password, String url)
+ throws JMSException, ClassNotFoundException, InstantiationException,
+ IllegalAccessException, InvocationTargetException, NoSuchMethodException {
+
+ @SuppressWarnings("unchecked")
+ Class<ConnectionFactory> clazz = (Class<ConnectionFactory>)
+ JMSMessageConsumer.class.getClassLoader().loadClass(implementation);
+
+ ConnectionFactory connectionFactory = clazz.getConstructor(
+ String.class, String.class, String.class).newInstance(userName,
+ password, url);
+
+ return connectionFactory.createConnection();
+ }
+
+ @Override
+ public String toString() {
+ return topicName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageProducer.java
----------------------------------------------------------------------
diff --git a/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageProducer.java b/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageProducer.java
new file mode 100644
index 0000000..6c9859c
--- /dev/null
+++ b/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageProducer.java
@@ -0,0 +1,376 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.messaging;
+
+import org.apache.falcon.entity.v0.SchemaHelper;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.Topic;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.InvocationTargetException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Message producer used in the workflow to send a message to the queue/topic.
+ */
+public class JMSMessageProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JMSMessageProducer.class);
+
+ /**
+ * Message messageType.
+ */
+ public enum MessageType {FALCON, USER}
+
+ public static final String FALCON_TOPIC_PREFIX = "FALCON.";
+ public static final String ENTITY_TOPIC_NAME = "ENTITY.TOPIC";
+
+ private static final long DEFAULT_TTL = 3 * 24 * 60 * 60 * 1000;
+
+ private static final WorkflowExecutionArgs[] FALCON_FILTER = {
+ WorkflowExecutionArgs.NOMINAL_TIME,
+ WorkflowExecutionArgs.ENTITY_NAME,
+ WorkflowExecutionArgs.OPERATION,
+ WorkflowExecutionArgs.LOG_DIR,
+ WorkflowExecutionArgs.STATUS,
+ WorkflowExecutionArgs.CONTEXT_FILE,
+ WorkflowExecutionArgs.TIMESTAMP,
+ };
+
+ private static final WorkflowExecutionArgs[] USER_FILTER = {
+ WorkflowExecutionArgs.CLUSTER_NAME,
+ WorkflowExecutionArgs.ENTITY_NAME,
+ WorkflowExecutionArgs.ENTITY_TYPE,
+ WorkflowExecutionArgs.NOMINAL_TIME,
+ WorkflowExecutionArgs.OPERATION,
+
+ WorkflowExecutionArgs.FEED_NAMES,
+ WorkflowExecutionArgs.FEED_INSTANCE_PATHS,
+
+ WorkflowExecutionArgs.WORKFLOW_ID,
+ WorkflowExecutionArgs.WORKFLOW_USER,
+ WorkflowExecutionArgs.RUN_ID,
+ WorkflowExecutionArgs.STATUS,
+ WorkflowExecutionArgs.TIMESTAMP,
+ WorkflowExecutionArgs.LOG_FILE,
+ };
+
+
+ private final WorkflowExecutionContext context;
+ private final MessageType messageType;
+
+ protected JMSMessageProducer(WorkflowExecutionContext context, MessageType messageType) {
+ this.context = context;
+ this.messageType = messageType;
+ }
+
+ public boolean isFalconEntityTopic() {
+ return messageType == MessageType.FALCON;
+ }
+
+ // convention over configuration
+ public String getTopicName() {
+ String topicNameValue = context.getValue(WorkflowExecutionArgs.TOPIC_NAME);
+ return topicNameValue != null
+ ? topicNameValue // return if user has set a topic
+ : FALCON_TOPIC_PREFIX // else falcon entity topic or user = FALCON.$entity_name
+ + (messageType == MessageType.FALCON ? ENTITY_TOPIC_NAME : context.getEntityName());
+ }
+
+ public String getBrokerImplClass() {
+ return messageType == MessageType.FALCON
+ ? context.getValue(WorkflowExecutionArgs.BRKR_IMPL_CLASS)
+ : context.getValue(WorkflowExecutionArgs.USER_BRKR_IMPL_CLASS);
+ }
+
+ public String getBrokerUrl() {
+ return messageType == MessageType.FALCON
+ ? context.getValue(WorkflowExecutionArgs.BRKR_URL)
+ : context.getValue(WorkflowExecutionArgs.USER_BRKR_URL);
+ }
+
+ private long getBrokerTTL() {
+ long messageTTL = DEFAULT_TTL;
+
+ try {
+ long messageTTLinMins = Long.parseLong(context.getValue(WorkflowExecutionArgs.BRKR_TTL));
+ messageTTL = messageTTLinMins * 60 * 1000;
+ } catch (NumberFormatException e) {
+ LOG.error("Error in parsing broker.ttl, setting TTL to: {} milli-seconds", DEFAULT_TTL);
+ }
+
+ return messageTTL;
+ }
+
+ public static MessageBuilder builder(WorkflowExecutionContext context) {
+ return new MessageBuilder(context);
+ }
+
+ /**
+ * Builder for JMSMessageProducer.
+ */
+ public static final class MessageBuilder {
+ private final WorkflowExecutionContext context;
+ private MessageType type;
+
+ private MessageBuilder(WorkflowExecutionContext context) {
+ this.context = context;
+ }
+
+ public MessageBuilder type(MessageType aMessageType) {
+ this.type = aMessageType;
+ return this;
+ }
+
+ public JMSMessageProducer build() {
+ if (type == null) {
+ throw new IllegalArgumentException("Message messageType needs to be set.");
+ }
+ return new JMSMessageProducer(context, type);
+ }
+ }
+
+ /**
+ * Accepts a Message to be send to JMS topic, creates a new
+ * Topic based on topic name if it does not exist or else
+ * existing topic with the same name is used to send the message.
+ */
+ public int sendMessage() throws JMSException {
+ List<Map<String, String>> messageList = buildMessageList();
+
+ if (messageList.isEmpty()) {
+ LOG.warn("No operation on output feed");
+ return 0;
+ }
+
+ Connection connection = null;
+ try {
+ connection = createAndStartConnection(getBrokerImplClass(), "", "", getBrokerUrl());
+
+ for (Map<String, String> message : messageList) {
+ LOG.info("Sending message: {}", message);
+ sendMessage(connection, message);
+ }
+ } catch (JMSException e) {
+ LOG.error("Error in getConnection:", e);
+ } catch (Exception e) {
+ LOG.error("Error in getConnection:", e);
+ } finally {
+ closeQuietly(connection);
+ }
+
+ return 0;
+ }
+
+ private List<Map<String, String>> buildMessageList() {
+ WorkflowExecutionArgs[] fileredArgs = messageType == MessageType.FALCON
+ ? FALCON_FILTER : USER_FILTER;
+
+ String[] feedNames = getFeedNames();
+ if (feedNames == null) {
+ return Collections.emptyList();
+ }
+
+ String[] feedPaths;
+ try {
+ feedPaths = getFeedPaths();
+ } catch (IOException e) {
+ LOG.error("Error getting instance paths: ", e);
+ throw new RuntimeException(e);
+ }
+
+ List<Map<String, String>> messages = new ArrayList<Map<String, String>>(feedPaths.length);
+ for (int i = 0; i < feedPaths.length; i++) {
+ Map<String, String> message = buildMessage(fileredArgs);
+
+ // override default values
+ if (context.getEntityType().equalsIgnoreCase("PROCESS")) {
+ change(message, WorkflowExecutionArgs.FEED_NAMES, feedNames[i]);
+ } else {
+ change(message, WorkflowExecutionArgs.FEED_NAMES,
+ message.get(WorkflowExecutionArgs.FEED_NAMES.getName()));
+ }
+
+ change(message, WorkflowExecutionArgs.FEED_INSTANCE_PATHS, feedPaths[i]);
+ convertDateFormat(message);
+ messages.add(message);
+ }
+
+ return messages;
+ }
+
+ private void sendMessage(Connection connection,
+ Map<String, String> message) throws JMSException {
+ Session session = null;
+ javax.jms.MessageProducer producer = null;
+ try {
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic entityTopic = session.createTopic(getTopicName());
+
+ producer = session.createProducer(entityTopic);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.setTimeToLive(getBrokerTTL());
+
+ producer.send(createMessage(session, message));
+ } finally {
+ if (producer != null) {
+ producer.close();
+ }
+
+ if (session != null) {
+ session.close();
+ }
+ }
+ }
+
+ public Message createMessage(Session session,
+ Map<String, String> message) throws JMSException {
+ MapMessage mapMessage = session.createMapMessage();
+
+ for (Map.Entry<String, String> entry : message.entrySet()) {
+ mapMessage.setString(entry.getKey(), entry.getValue());
+ }
+
+ return mapMessage;
+ }
+
+ public void change(Map<String, String> message, WorkflowExecutionArgs key, String value) {
+ message.remove(key.getName());
+ message.put(key.getName(), value);
+ }
+
+ private String[] getFeedNames() {
+ String feedNameStr = context.getOutputFeedNames();
+ if (isFalconEntityTopic()) {
+ return new String[]{feedNameStr};
+ }
+
+ if (feedNameStr.equals("null")) {
+ return null;
+ }
+
+ return context.getOutputFeedNamesList();
+ }
+
+ private String[] getFeedPaths() throws IOException {
+
+ if (isFalconEntityTopic()) {
+ LOG.debug("Returning instance paths for Falcon Topic: " + context.getOutputFeedInstancePaths());
+ return new String[]{context.getOutputFeedInstancePaths(), };
+ }
+
+ WorkflowExecutionContext.EntityOperations operation = context.getOperation();
+ if (operation == WorkflowExecutionContext.EntityOperations.GENERATE
+ || operation == WorkflowExecutionContext.EntityOperations.REPLICATE) {
+ LOG.debug("Returning instance paths: " + context.getOutputFeedInstancePaths());
+ return context.getOutputFeedInstancePathsList();
+ }
+
+ // else case of feed retention
+ Path logFile = new Path(context.getLogFile());
+ FileSystem fs = FileSystem.get(logFile.toUri(), new Configuration());
+
+ if (!fs.exists(logFile)) {
+ // Evictor Failed without deleting a single path
+ return new String[0];
+ }
+
+ ByteArrayOutputStream writer = new ByteArrayOutputStream();
+ InputStream instance = fs.open(logFile);
+ IOUtils.copyBytes(instance, writer, 4096, true);
+ String[] instancePaths = writer.toString().split("=");
+ fs.delete(logFile, true);
+ LOG.info("Deleted feed instance paths file:" + logFile);
+ if (instancePaths.length == 1) {
+ LOG.debug("Returning 0 instance paths for feed ");
+ return new String[0];
+ } else {
+ LOG.debug("Returning instance paths for feed " + instancePaths[1]);
+ return instancePaths[1].split(",");
+ }
+ }
+
+ private Map<String, String> buildMessage(final WorkflowExecutionArgs[] filter) {
+ Map<String, String> message = new HashMap<String, String>(filter.length);
+ for (WorkflowExecutionArgs arg : filter) {
+ message.put(arg.getName(), context.getValue(arg));
+ }
+
+ return message;
+ }
+
+ public void convertDateFormat(Map<String, String> message) {
+ String date = message.get(WorkflowExecutionArgs.NOMINAL_TIME.getName());
+ change(message, WorkflowExecutionArgs.NOMINAL_TIME,
+ SchemaHelper.formatDateUTCToISO8601(date, "yyyy-MM-dd-HH-mm"));
+
+ date = message.get(WorkflowExecutionArgs.TIMESTAMP.getName());
+ change(message, WorkflowExecutionArgs.TIMESTAMP,
+ SchemaHelper.formatDateUTCToISO8601(date, "yyyy-MM-dd-HH-mm"));
+ }
+
+ @SuppressWarnings("unchecked")
+ private Connection createAndStartConnection(String implementation, String userName,
+ String password, String url)
+ throws JMSException, ClassNotFoundException, InstantiationException,
+ IllegalAccessException, InvocationTargetException, NoSuchMethodException {
+
+ Class<ConnectionFactory> clazz = (Class<ConnectionFactory>)
+ JMSMessageProducer.class.getClassLoader().loadClass(implementation);
+
+ ConnectionFactory connectionFactory = clazz
+ .getConstructor(String.class, String.class, String.class)
+ .newInstance(userName, password, url);
+
+ Connection connection = connectionFactory.createConnection();
+ connection.start();
+
+ return connection;
+ }
+
+ private void closeQuietly(Connection connection) {
+ try {
+ if (connection != null) {
+ connection.close();
+ }
+ } catch (JMSException e) {
+ LOG.error("Error in closing connection:", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/messaging/src/main/java/org/apache/falcon/messaging/MessageProducer.java
----------------------------------------------------------------------
diff --git a/messaging/src/main/java/org/apache/falcon/messaging/MessageProducer.java b/messaging/src/main/java/org/apache/falcon/messaging/MessageProducer.java
deleted file mode 100644
index ccac921..0000000
--- a/messaging/src/main/java/org/apache/falcon/messaging/MessageProducer.java
+++ /dev/null
@@ -1,180 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-
-package org.apache.falcon.messaging;
-
-import org.apache.commons.cli.*;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
-import org.apache.hadoop.conf.Configured;
-import org.apache.hadoop.util.Tool;
-import org.apache.hadoop.util.ToolRunner;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.jms.*;
-import java.lang.reflect.InvocationTargetException;
-
-/**
- * Message producer used in the workflow to send a message to the queue/topic.
- */
-public class MessageProducer extends Configured implements Tool {
-
- private Connection connection;
- private static final Logger LOG = LoggerFactory.getLogger(MessageProducer.class);
- private static final long DEFAULT_TTL = 3 * 24 * 60 * 60 * 1000;
-
- /**
- * @param entityInstanceMessage - Accepts a Message to be send to JMS topic, creates a new
- * Topic based on topic name if it does not exist or else
- * existing topic with the same name is used to send the message.
- * @throws JMSException
- */
- protected void sendMessage(EntityInstanceMessage entityInstanceMessage)
- throws JMSException {
-
- Session session = connection.createSession(false,
- Session.AUTO_ACKNOWLEDGE);
- Topic entityTopic = session.createTopic(entityInstanceMessage
- .getTopicName());
- javax.jms.MessageProducer producer = session
- .createProducer(entityTopic);
- producer.setDeliveryMode(DeliveryMode.PERSISTENT);
- long messageTTL = DEFAULT_TTL;
- try {
- long messageTTLinMins = Long.parseLong(entityInstanceMessage
- .getBrokerTTL());
- messageTTL = messageTTLinMins * 60 * 1000;
- } catch (NumberFormatException e) {
- LOG.error("Error in parsing broker.ttl, setting TTL to: {} milli-seconds", DEFAULT_TTL);
- }
- producer.setTimeToLive(messageTTL);
- producer.send(new EntityInstanceMessageCreator(entityInstanceMessage)
- .createMessage(session));
- }
-
- public static void main(String[] args) throws Exception {
- ToolRunner.run(new MessageProducer(), args);
- }
-
- private void createAndStartConnection(String implementation, String userName,
- String password, String url)
- throws JMSException, ClassNotFoundException, InstantiationException,
- IllegalAccessException, InvocationTargetException, NoSuchMethodException {
-
- Class<ConnectionFactory> clazz = (Class<ConnectionFactory>) MessageProducer.class
- .getClassLoader().loadClass(implementation);
-
- ConnectionFactory connectionFactory = clazz.getConstructor(
- String.class, String.class, String.class).newInstance(userName,
- password, url);
-
- connection = connectionFactory.createConnection();
- connection.start();
- }
-
- private static CommandLine getCommand(String[] arguments)
- throws ParseException {
-
- Options options = new Options();
- addOption(options, new Option(ARG.brokerImplClass.getArgName(), true,
- "message broker Implementation class"));
- addOption(options, new Option(ARG.brokerTTL.getArgName(), true,
- "message time-to-live"));
- addOption(options, new Option(ARG.brokerUrl.getArgName(), true,
- "message broker url"));
- addOption(options, new Option(ARG.entityName.getArgName(), true,
- "name of the entity"));
- addOption(options, new Option(ARG.entityType.getArgName(), true,
- "type of the entity"));
- addOption(options, new Option(ARG.feedInstancePaths.getArgName(),
- true, "feed instance paths"));
- addOption(options, new Option(ARG.feedNames.getArgName(), true,
- "feed names"));
- addOption(options, new Option(ARG.logFile.getArgName(), true,
- "log file path"));
- addOption(options, new Option(ARG.nominalTime.getArgName(), true,
- "instance time"));
- addOption(options, new Option(ARG.operation.getArgName(), true,
- "operation like generate, delete, archive"));
- addOption(options, new Option(ARG.runId.getArgName(), true,
- "current run-id of the instance"));
- addOption(options, new Option(ARG.status.getArgName(), true,
- "status of workflow instance"));
- addOption(options, new Option(ARG.timeStamp.getArgName(), true,
- "current timestamp"));
- addOption(options, new Option(ARG.topicName.getArgName(), true,
- "name of the topic to be used to send message"));
- addOption(options, new Option(ARG.workflowId.getArgName(), true,
- "workflow id"));
- addOption(options, new Option(ARG.cluster.getArgName(), true,
- "cluster name"));
- addOption(options, new Option(ARG.workflowUser.getArgName(), true,
- "workflow user id"), false);
- addOption(options, new Option(ARG.logDir.getArgName(), true,
- "log dir where job logs are copied"), false);
-
- return new GnuParser().parse(options, arguments);
- }
-
- private static void addOption(Options options, Option opt) {
- addOption(options, opt, true);
- }
-
- private static void addOption(Options options, Option opt, boolean isRequired) {
- opt.setRequired(isRequired);
- options.addOption(opt);
- }
-
- @Override
- public int run(String[] args) throws Exception {
- CommandLine cmd;
- try {
- cmd = getCommand(args);
- } catch (ParseException e) {
- throw new Exception("Unable to parse arguments: ", e);
- }
- EntityInstanceMessage[] entityInstanceMessage = EntityInstanceMessage
- .getMessages(cmd);
- if (entityInstanceMessage == null || entityInstanceMessage.length == 0) {
- LOG.warn("No operation on output feed");
- return 0;
- }
-
- try {
- createAndStartConnection(cmd.getOptionValue(ARG.brokerImplClass.name()), "",
- "", cmd.getOptionValue(ARG.brokerUrl.name()));
- for (EntityInstanceMessage message : entityInstanceMessage) {
- LOG.info("Sending message: {}", message.getKeyValueMap());
- sendMessage(message);
- }
- } catch (JMSException e) {
- LOG.error("Error in getConnection", e);
- } catch (Exception e) {
- LOG.error("Error in getConnection", e);
- } finally {
- try {
- if (connection != null) {
- connection.close();
- }
- } catch (JMSException e) {
- LOG.error("Error in closing connection", e);
- }
- }
- return 0;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/messaging/src/test/java/org/apache/falcon/messaging/FalconTopicProducerTest.java
----------------------------------------------------------------------
diff --git a/messaging/src/test/java/org/apache/falcon/messaging/FalconTopicProducerTest.java b/messaging/src/test/java/org/apache/falcon/messaging/FalconTopicProducerTest.java
deleted file mode 100644
index 3f0c664..0000000
--- a/messaging/src/test/java/org/apache/falcon/messaging/FalconTopicProducerTest.java
+++ /dev/null
@@ -1,215 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-package org.apache.falcon.messaging;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
-import org.testng.Assert;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.Test;
-
-import javax.jms.*;
-
-/**
- * Test for falcon topic message producer.
- */
-public class FalconTopicProducerTest {
-
- 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 TOPIC_NAME = "FALCON.ENTITY.TOPIC";
- private static final String SECONDARY_TOPIC_NAME = "FALCON.ENTITY.SEC.TOPIC";
- private BrokerService broker;
- private List<MapMessage> mapMessages;
-
- private volatile AssertionError error;
-
- @BeforeClass
- public void setup() throws Exception {
- broker = new BrokerService();
- broker.addConnector(BROKER_URL);
- broker.setDataDirectory("target/activemq");
- broker.setBrokerName("localhost");
- broker.start();
- }
-
- @AfterClass
- public void tearDown() throws Exception {
- broker.deleteAllMessages();
- broker.stop();
- }
-
- @Test
- public void testWithFeedOutputPaths() throws Exception {
- List<String> args = createCommonArgs();
- List<String> newArgs = new ArrayList<String>(Arrays.asList(
- "-" + ARG.entityName.getArgName(), "agg-coord",
- "-" + ARG.feedNames.getArgName(), "click-logs,raw-logs",
- "-" + ARG.feedInstancePaths.getArgName(),
- "/click-logs/10/05/05/00/20,/raw-logs/10/05/05/00/20",
- "-" + ARG.topicName.getArgName(), TOPIC_NAME));
- args.addAll(newArgs);
- List<String[]> messages = new ArrayList<String[]>();
- messages.add(args.toArray(new String[args.size()]));
- testProcessMessageCreator(messages, TOPIC_NAME);
- for (MapMessage m : mapMessages) {
- assertMessage(m);
- Assert.assertTrue((m.getString(ARG.feedNames.getArgName())
- .equals("click-logs,raw-logs")));
- Assert.assertTrue(m
- .getString(ARG.feedInstancePaths.getArgName())
- .equals("/click-logs/10/05/05/00/20,/raw-logs/10/05/05/00/20"));
- }
- }
-
- @Test
- public void testWithEmptyFeedOutputPaths() throws Exception {
- List<String> args = createCommonArgs();
- List<String> newArgs = new ArrayList<String>(Arrays.asList(
- "-" + ARG.entityName.getArgName(), "agg-coord",
- "-" + ARG.feedNames.getArgName(), "null",
- "-" + ARG.feedInstancePaths.getArgName(), "null",
- "-" + ARG.topicName.getArgName(), TOPIC_NAME));
- args.addAll(newArgs);
- List<String[]> messages = new ArrayList<String[]>();
- messages.add(args.toArray(new String[args.size()]));
- testProcessMessageCreator(messages, TOPIC_NAME);
- for (MapMessage m : mapMessages) {
- assertMessage(m);
- assertMessage(m);
- Assert.assertTrue(m.getString(ARG.feedNames.getArgName()).equals(
- "null"));
- Assert.assertTrue(m.getString(ARG.feedInstancePaths.getArgName())
- .equals("null"));
- }
- }
-
- @Test
- public void testConsumerWithMultipleTopics() throws Exception {
- List<String[]> messages = new ArrayList<String[]>();
- List<String> args = createCommonArgs();
- List<String> newArgs = new ArrayList<String>(Arrays.asList(
- "-" + ARG.entityName.getArgName(), "agg-coord",
- "-" + ARG.feedNames.getArgName(), "raw-logs",
- "-" + ARG.feedInstancePaths.getArgName(),
- "/raw-logs/10/05/05/00/20",
- "-" + ARG.topicName.getArgName(), TOPIC_NAME));
- args.addAll(newArgs);
- messages.add(args.toArray(new String[args.size()]));
-
- args = createCommonArgs();
- newArgs = new ArrayList<String>(Arrays.asList(
- "-" + ARG.entityName.getArgName(), "agg-coord",
- "-" + ARG.feedNames.getArgName(), "click-logs",
- "-" + ARG.feedInstancePaths.getArgName(),
- "/click-logs/10/05/05/00/20",
- "-" + ARG.topicName.getArgName(), SECONDARY_TOPIC_NAME));
- args.addAll(newArgs);
- messages.add(args.toArray(new String[args.size()]));
-
- testProcessMessageCreator(messages, TOPIC_NAME+","+SECONDARY_TOPIC_NAME);
- Assert.assertEquals(mapMessages.size(), 2);
- for (MapMessage m : mapMessages) {
- assertMessage(m);
- }
- }
-
- private List<String> createCommonArgs() {
- return new ArrayList<String>(Arrays.asList(
- "-" + ARG.workflowId.getArgName(), "workflow-01-00",
- "-" + ARG.workflowUser.getArgName(), "falcon",
- "-" + ARG.runId.getArgName(), "1",
- "-" + ARG.nominalTime.getArgName(), "2011-01-01-01-00",
- "-" + ARG.timeStamp.getArgName(), "2012-01-01-01-00",
- "-" + ARG.brokerUrl.getArgName(), BROKER_URL,
- "-" + ARG.brokerImplClass.getArgName(), (BROKER_IMPL_CLASS),
- "-" + ARG.entityType.getArgName(), ("process"),
- "-" + ARG.operation.getArgName(), ("GENERATE"),
- "-" + ARG.logFile.getArgName(), ("/logFile"),
- "-" + ARG.status.getArgName(), ("SUCCEEDED"),
- "-" + ARG.brokerTTL.getArgName(), "10",
- "-" + ARG.cluster.getArgName(), "corp"));
- }
-
- private void testProcessMessageCreator(final List<String[]> messages,
- final String topicsToListen) throws Exception {
-
- Thread t = new Thread() {
- @Override
- public void run() {
- try {
- consumer(messages.size(), topicsToListen);
- } catch (AssertionError e) {
- error = e;
- } catch (Exception ignore) {
- error = null;
- }
- }
- };
- t.start();
- Thread.sleep(100);
- for (String[] message : messages) {
- new MessageProducer().run(message);
- }
- t.join();
- if (error != null) {
- throw error;
- }
- }
-
- private void consumer(int size, String topicsToListen) throws Exception {
- ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
- BROKER_URL);
- Connection connection = connectionFactory.createConnection();
- connection.start();
-
- Session session = connection.createSession(false,
- Session.AUTO_ACKNOWLEDGE);
- Destination destination = session.createTopic(topicsToListen);
- MessageConsumer consumer = session.createConsumer(destination);
- mapMessages = new ArrayList<MapMessage>();
- for (int i=0; i<size; i++) {
- MapMessage m = (MapMessage) consumer.receive();
- mapMessages.add(m);
- System.out.println("Consumed: " + m.toString());
- }
-
- connection.close();
- }
-
- private void assertMessage(MapMessage m) throws JMSException {
- Assert.assertEquals(m.getString(ARG.entityName.getArgName()),
- "agg-coord");
- Assert.assertEquals(m.getString(ARG.workflowId.getArgName()),
- "workflow-01-00");
- Assert.assertEquals(m.getString(ARG.workflowUser.getArgName()),
- "falcon");
- Assert.assertEquals(m.getString(ARG.runId.getArgName()), "1");
- Assert.assertEquals(m.getString(ARG.nominalTime.getArgName()),
- "2011-01-01T01:00Z");
- Assert.assertEquals(m.getString(ARG.timeStamp.getArgName()),
- "2012-01-01T01:00Z");
- Assert.assertEquals(m.getString(ARG.status.getArgName()), "SUCCEEDED");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/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 57ccdc5..9119624 100644
--- a/messaging/src/test/java/org/apache/falcon/messaging/FeedProducerTest.java
+++ b/messaging/src/test/java/org/apache/falcon/messaging/FeedProducerTest.java
@@ -33,7 +33,8 @@ import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.falcon.cluster.util.EmbeddedCluster;
-import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -68,24 +69,29 @@ public class FeedProducerTest {
logFile = new Path(conf.get("fs.default.name"),
"/falcon/feed/agg-logs/instance-2012-01-01-10-00.csv");
- args = new String[]{"-" + ARG.entityName.getArgName(), TOPIC_NAME,
- "-" + ARG.feedNames.getArgName(), "click-logs",
- "-" + ARG.feedInstancePaths.getArgName(),
- "/click-logs/10/05/05/00/20",
- "-" + ARG.workflowId.getArgName(), "workflow-01-00",
- "-" + ARG.workflowUser.getArgName(), "falcon",
- "-" + ARG.runId.getArgName(), "1",
- "-" + ARG.nominalTime.getArgName(), "2011-01-01-01-00",
- "-" + ARG.timeStamp.getArgName(), "2012-01-01-01-00",
- "-" + ARG.brokerUrl.getArgName(), BROKER_URL,
- "-" + ARG.brokerImplClass.getArgName(), (BROKER_IMPL_CLASS),
- "-" + ARG.entityType.getArgName(), ("FEED"),
- "-" + ARG.operation.getArgName(), ("DELETE"),
- "-" + ARG.logFile.getArgName(), (logFile.toString()),
- "-" + ARG.topicName.getArgName(), (TOPIC_NAME),
- "-" + ARG.status.getArgName(), ("SUCCEEDED"),
- "-" + ARG.brokerTTL.getArgName(), "10",
- "-" + ARG.cluster.getArgName(), "corp", };
+ args = new String[] {
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), TOPIC_NAME,
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), "click-logs",
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(),
+ "/click-logs/10/05/05/00/20",
+ "-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
+ "-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), "falcon",
+ "-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
+ "-" + WorkflowExecutionArgs.NOMINAL_TIME.getName(), "2011-01-01-01-00",
+ "-" + WorkflowExecutionArgs.TIMESTAMP.getName(), "2012-01-01-01-00",
+ "-" + WorkflowExecutionArgs.BRKR_URL.getName(), BROKER_URL,
+ "-" + WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), BROKER_IMPL_CLASS,
+ "-" + WorkflowExecutionArgs.USER_BRKR_URL.getName(), BROKER_URL,
+ "-" + WorkflowExecutionArgs.USER_BRKR_IMPL_CLASS.getName(), BROKER_IMPL_CLASS,
+ "-" + WorkflowExecutionArgs.ENTITY_TYPE.getName(), "FEED",
+ "-" + WorkflowExecutionArgs.OPERATION.getName(), "DELETE",
+ "-" + WorkflowExecutionArgs.LOG_FILE.getName(), logFile.toString(),
+ "-" + WorkflowExecutionArgs.LOG_DIR.getName(), "/falcon/feed/agg-logs/",
+ "-" + WorkflowExecutionArgs.TOPIC_NAME.getName(), TOPIC_NAME,
+ "-" + WorkflowExecutionArgs.STATUS.getName(), "SUCCEEDED",
+ "-" + WorkflowExecutionArgs.BRKR_TTL.getName(), "10",
+ "-" + WorkflowExecutionArgs.CLUSTER_NAME.getName(), "corp",
+ };
broker = new BrokerService();
broker.addConnector(BROKER_URL);
@@ -121,7 +127,11 @@ public class FeedProducerTest {
InputStream in = new ByteArrayInputStream(("instancePaths=").getBytes());
IOUtils.copyBytes(in, out, conf);
- new MessageProducer().run(this.args);
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(
+ args, WorkflowExecutionContext.Type.POST_PROCESSING);
+ JMSMessageProducer jmsMessageProducer = JMSMessageProducer.builder(context)
+ .type(JMSMessageProducer.MessageType.USER).build();
+ jmsMessageProducer.sendMessage();
}
private void testProcessMessageCreator() throws Exception {
@@ -140,7 +150,13 @@ public class FeedProducerTest {
};
t.start();
Thread.sleep(100);
- new MessageProducer().run(this.args);
+
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(
+ args, WorkflowExecutionContext.Type.POST_PROCESSING);
+ JMSMessageProducer jmsMessageProducer = JMSMessageProducer.builder(context)
+ .type(JMSMessageProducer.MessageType.USER).build();
+ jmsMessageProducer.sendMessage();
+
t.join();
if (error != null) {
throw error;
@@ -148,13 +164,11 @@ public class FeedProducerTest {
}
private void consumer() throws JMSException {
- ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
- BROKER_URL);
+ ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
Connection connection = connectionFactory.createConnection();
connection.start();
- Session session = connection.createSession(false,
- Session.AUTO_ACKNOWLEDGE);
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic(TOPIC_NAME);
MessageConsumer consumer = session.createConsumer(destination);
@@ -165,7 +179,7 @@ public class FeedProducerTest {
}
System.out.println("Consumed: " + m.toString());
assertMessage(m);
- Assert.assertEquals(m.getString(ARG.feedInstancePaths.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName()),
"/falcon/feed/agg-logs/path1/2010/10/10/20");
for (m = null; m == null;) {
@@ -173,7 +187,7 @@ public class FeedProducerTest {
}
System.out.println("Consumed: " + m.toString());
assertMessage(m);
- Assert.assertEquals(m.getString(ARG.feedInstancePaths.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName()),
"/falcon/feed/agg-logs/path1/2010/10/10/21");
for (m = null; m == null;) {
@@ -181,7 +195,7 @@ public class FeedProducerTest {
}
System.out.println("Consumed: " + m.toString());
assertMessage(m);
- Assert.assertEquals(m.getString(ARG.feedInstancePaths.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName()),
"/falcon/feed/agg-logs/path1/2010/10/10/22");
for (m = null; m == null;) {
@@ -189,25 +203,25 @@ public class FeedProducerTest {
}
System.out.println("Consumed: " + m.toString());
assertMessage(m);
- Assert.assertEquals(m.getString(ARG.feedInstancePaths.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName()),
"/falcon/feed/agg-logs/path1/2010/10/10/23");
connection.close();
}
private void assertMessage(MapMessage m) throws JMSException {
- Assert.assertEquals(m.getString(ARG.entityName.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.ENTITY_NAME.getName()),
TOPIC_NAME);
- Assert.assertEquals(m.getString(ARG.operation.getArgName()), "DELETE");
- Assert.assertEquals(m.getString(ARG.workflowId.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.OPERATION.getName()), "DELETE");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.WORKFLOW_ID.getName()),
"workflow-01-00");
- Assert.assertEquals(m.getString(ARG.workflowUser.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.WORKFLOW_USER.getName()),
"falcon");
- Assert.assertEquals(m.getString(ARG.runId.getArgName()), "1");
- Assert.assertEquals(m.getString(ARG.nominalTime.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.RUN_ID.getName()), "1");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.NOMINAL_TIME.getName()),
"2011-01-01T01:00Z");
- Assert.assertEquals(m.getString(ARG.timeStamp.getArgName()),
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.TIMESTAMP.getName()),
"2012-01-01T01:00Z");
- Assert.assertEquals(m.getString(ARG.status.getArgName()), "SUCCEEDED");
+ Assert.assertEquals(m.getString(WorkflowExecutionArgs.STATUS.getName()), "SUCCEEDED");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/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
new file mode 100644
index 0000000..694d488
--- /dev/null
+++ b/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageConsumerTest.java
@@ -0,0 +1,156 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.messaging;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.falcon.FalconException;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
+import org.apache.falcon.workflow.WorkflowJobEndNotificationService;
+import org.mortbay.log.Log;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.Session;
+import javax.jms.Topic;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Test for FalconTopicSubscriber.
+ */
+public class JMSMessageConsumerTest {
+
+ private static final String BROKER_URL = "vm://localhost";
+ private static final String BROKER_IMPL_CLASS = "org.apache.activemq.ActiveMQConnectionFactory";
+ private static final String TOPIC_NAME = "FALCON.ENTITY.TOPIC";
+ private static final String SECONDARY_TOPIC_NAME = "FALCON.ENTITY.SEC.TOPIC";
+ private BrokerService broker;
+
+ @BeforeClass
+ public void setup() throws Exception {
+ broker = new BrokerService();
+ broker.addConnector(BROKER_URL);
+ broker.setDataDirectory("target/activemq");
+ broker.setBrokerName("localhost");
+ broker.start();
+ }
+
+ public void sendMessages(String topic) throws JMSException, FalconException, IOException {
+ ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
+ Connection connection = connectionFactory.createConnection();
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic destination = session.createTopic(topic);
+ javax.jms.MessageProducer producer = session.createProducer(destination);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+ for (int i = 0; i < 3; i++) {
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(
+ getMockFalconMessage(i), WorkflowExecutionContext.Type.POST_PROCESSING);
+ context.serialize(WorkflowExecutionContext.getFilePath("/tmp/log", "process1"));
+
+ MapMessage message = session.createMapMessage();
+ for (Map.Entry<WorkflowExecutionArgs, String> entry : context.entrySet()) {
+ message.setString(entry.getKey().getName(), entry.getValue());
+ }
+
+ Log.debug("Sending:" + message);
+ producer.send(message);
+ }
+
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(
+ getMockFalconMessage(15), WorkflowExecutionContext.Type.POST_PROCESSING);
+ context.serialize(WorkflowExecutionContext.getFilePath("/tmp/log", "process1"));
+
+ MapMessage mapMessage = session.createMapMessage();
+ for (Map.Entry<WorkflowExecutionArgs, String> entry : context.entrySet()) {
+ mapMessage.setString(entry.getKey().getName(), entry.getValue());
+ }
+
+ Log.debug("Sending:" + mapMessage);
+ producer.send(mapMessage);
+ }
+
+ private String[] getMockFalconMessage(int i) {
+ Map<String, String> message = new HashMap<String, String>();
+ message.put(WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), BROKER_IMPL_CLASS);
+ message.put(WorkflowExecutionArgs.BRKR_URL.getName(), BROKER_URL);
+ message.put(WorkflowExecutionArgs.CLUSTER_NAME.getName(), "cluster1");
+ message.put(WorkflowExecutionArgs.ENTITY_NAME.getName(), "process1");
+ message.put(WorkflowExecutionArgs.ENTITY_TYPE.getName(), "PROCESS");
+ message.put(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(),
+ "/clicks/hour/00/0" + i);
+ message.put(WorkflowExecutionArgs.FEED_NAMES.getName(), "clicks");
+ message.put(WorkflowExecutionArgs.LOG_FILE.getName(), "/logfile");
+ message.put(WorkflowExecutionArgs.LOG_DIR.getName(), "/tmp/log");
+ message.put(WorkflowExecutionArgs.NOMINAL_TIME.getName(), "2012-10-10-10-10");
+ message.put(WorkflowExecutionArgs.OPERATION.getName(), "GENERATE");
+ message.put(WorkflowExecutionArgs.RUN_ID.getName(), "0");
+ message.put(WorkflowExecutionArgs.TIMESTAMP.getName(), "2012-10-10-10-1" + i);
+ message.put(WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-" + i);
+ message.put(WorkflowExecutionArgs.TOPIC_NAME.getName(), TOPIC_NAME);
+ message.put(WorkflowExecutionArgs.STATUS.getName(), i != 15 ? "SUCCEEDED" : "FAILED");
+ message.put(WorkflowExecutionArgs.WORKFLOW_USER.getName(), "falcon");
+
+ String[] args = new String[message.size() * 2];
+ int index = 0;
+ for (Map.Entry<String, String> entry : message.entrySet()) {
+ args[index++] = "-" + entry.getKey();
+ args[index++] = entry.getValue();
+ }
+
+ return args;
+ }
+
+ @Test
+ public void testSubscriber() {
+ try {
+ //Comma separated topics are supported in startup properties
+ JMSMessageConsumer subscriber = new JMSMessageConsumer(BROKER_IMPL_CLASS, "", "",
+ BROKER_URL, TOPIC_NAME+","+SECONDARY_TOPIC_NAME, new WorkflowJobEndNotificationService());
+ subscriber.startSubscriber();
+ sendMessages(TOPIC_NAME);
+ Assert.assertEquals(broker.getAdminView().getTotalEnqueueCount(), 9);
+
+ sendMessages(SECONDARY_TOPIC_NAME);
+ Assert.assertEquals(broker.getAdminView().getTotalEnqueueCount(), 17);
+ Assert.assertEquals(broker.getAdminView().getTotalConsumerCount(), 2);
+ subscriber.closeSubscriber();
+ } catch (Exception e) {
+ Assert.fail("This should not have thrown an exception.", e);
+ }
+ }
+
+ @AfterClass
+ public void tearDown() throws Exception {
+ broker.deleteAllMessages();
+ broker.stop();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/23762e55/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
new file mode 100644
index 0000000..90efa3e
--- /dev/null
+++ b/messaging/src/test/java/org/apache/falcon/messaging/JMSMessageProducerTest.java
@@ -0,0 +1,218 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.messaging;
+
+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.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Test for falcon topic message producer.
+ */
+public class JMSMessageProducerTest {
+
+ 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 TOPIC_NAME = "FALCON.ENTITY.TOPIC";
+ private static final String SECONDARY_TOPIC_NAME = "FALCON.ENTITY.SEC.TOPIC";
+ private BrokerService broker;
+ private List<MapMessage> mapMessages;
+
+ private volatile AssertionError error;
+
+ @BeforeClass
+ public void setup() throws Exception {
+ broker = new BrokerService();
+ broker.addConnector(BROKER_URL);
+ broker.setDataDirectory("target/activemq");
+ broker.setBrokerName("localhost");
+ broker.start();
+ }
+
+ @AfterClass
+ public void tearDown() throws Exception {
+ broker.deleteAllMessages();
+ broker.stop();
+ }
+
+ @Test
+ public void testWithFeedOutputPaths() throws Exception {
+ List<String> args = createCommonArgs();
+ List<String> newArgs = new ArrayList<String>(Arrays.asList(
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), "agg-coord",
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), "click-logs,raw-logs",
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(),
+ "/click-logs/10/05/05/00/20,/raw-logs/10/05/05/00/20"));
+ args.addAll(newArgs);
+ List<String[]> messages = new ArrayList<String[]>();
+ messages.add(args.toArray(new String[args.size()]));
+ testProcessMessageCreator(messages, TOPIC_NAME);
+ for (MapMessage m : mapMessages) {
+ assertMessage(m);
+ Assert.assertTrue((m.getString(WorkflowExecutionArgs.FEED_NAMES.getName())
+ .equals("click-logs,raw-logs")));
+ Assert.assertTrue(m
+ .getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName())
+ .equals("/click-logs/10/05/05/00/20,/raw-logs/10/05/05/00/20"));
+ }
+ }
+
+ @Test
+ public void testWithEmptyFeedOutputPaths() throws Exception {
+ List<String> args = createCommonArgs();
+ List<String> newArgs = new ArrayList<String>(Arrays.asList(
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), "agg-coord",
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), "null",
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(), "null"));
+ args.addAll(newArgs);
+ List<String[]> messages = new ArrayList<String[]>();
+ messages.add(args.toArray(new String[args.size()]));
+ testProcessMessageCreator(messages, TOPIC_NAME);
+ for (MapMessage m : mapMessages) {
+ assertMessage(m);
+ assertMessage(m);
+ Assert.assertTrue(m.getString(WorkflowExecutionArgs.FEED_NAMES.getName()).equals(
+ "null"));
+ Assert.assertTrue(m.getString(WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName())
+ .equals("null"));
+ }
+ }
+
+ @Test
+ public void testConsumerWithMultipleTopics() throws Exception {
+ List<String[]> messages = new ArrayList<String[]>();
+ List<String> args = createCommonArgs();
+ List<String> newArgs = new ArrayList<String>(Arrays.asList(
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), "agg-coord",
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), "raw-logs",
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(),
+ "/raw-logs/10/05/05/00/20"));
+ args.addAll(newArgs);
+ messages.add(args.toArray(new String[args.size()]));
+
+ args = createCommonArgs();
+ newArgs = new ArrayList<String>(Arrays.asList(
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), "agg-coord",
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), "click-logs",
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(),
+ "/click-logs/10/05/05/00/20"));
+ args.addAll(newArgs);
+ messages.add(args.toArray(new String[args.size()]));
+
+ testProcessMessageCreator(messages, TOPIC_NAME+","+SECONDARY_TOPIC_NAME);
+ Assert.assertEquals(mapMessages.size(), 2);
+ for (MapMessage m : mapMessages) {
+ assertMessage(m);
+ }
+ }
+
+ private List<String> createCommonArgs() {
+ return new ArrayList<String>(Arrays.asList(
+ "-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
+ "-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), "falcon",
+ "-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
+ "-" + WorkflowExecutionArgs.NOMINAL_TIME.getName(), "2011-01-01-01-00",
+ "-" + WorkflowExecutionArgs.TIMESTAMP.getName(), "2012-01-01-01-00",
+ "-" + WorkflowExecutionArgs.BRKR_URL.getName(), BROKER_URL,
+ "-" + WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), (BROKER_IMPL_CLASS),
+ "-" + WorkflowExecutionArgs.ENTITY_TYPE.getName(), ("process"),
+ "-" + WorkflowExecutionArgs.OPERATION.getName(), ("GENERATE"),
+ "-" + WorkflowExecutionArgs.LOG_FILE.getName(), ("/logFile"),
+ "-" + WorkflowExecutionArgs.LOG_DIR.getName(), ("/tmp"),
+ "-" + WorkflowExecutionArgs.STATUS.getName(), ("SUCCEEDED"),
+ "-" + WorkflowExecutionArgs.BRKR_TTL.getName(), "10",
+ "-" + WorkflowExecutionArgs.CLUSTER_NAME.getName(), "corp"));
+ }
+
+ private void testProcessMessageCreator(final List<String[]> messages,
+ final String topicsToListen) throws Exception {
+ Thread t = new Thread() {
+ @Override
+ public void run() {
+ try {
+ consumer(messages.size(), topicsToListen);
+ } catch (AssertionError e) {
+ error = e;
+ } catch (Exception ignore) {
+ error = null;
+ }
+ }
+ };
+ t.start();
+ Thread.sleep(100);
+
+ for (String[] message : messages) {
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(
+ message, WorkflowExecutionContext.Type.POST_PROCESSING);
+ JMSMessageProducer jmsMessageProducer = JMSMessageProducer.builder(context)
+ .type(JMSMessageProducer.MessageType.FALCON).build();
+ jmsMessageProducer.sendMessage();
+ }
+
+ t.join();
+ if (error != null) {
+ throw error;
+ }
+ }
+
+ private void consumer(int size, String topicsToListen) throws Exception {
+ ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
+ Connection connection = connectionFactory.createConnection();
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createTopic(topicsToListen);
+ MessageConsumer consumer = session.createConsumer(destination);
+
+ mapMessages = new ArrayList<MapMessage>();
+ for (int i=0; i<size; i++) {
+ MapMessage m = (MapMessage) consumer.receive();
+ mapMessages.add(m);
+ System.out.println("Consumed: " + m.toString());
+ }
+
+ connection.close();
+ }
+
+ private void assertMessage(MapMessage message) throws JMSException {
+ Assert.assertEquals(message.getString(WorkflowExecutionArgs.ENTITY_NAME.getName()),
+ "agg-coord");
+ Assert.assertEquals(message.getString(WorkflowExecutionArgs.NOMINAL_TIME.getName()),
+ "2011-01-01T01:00Z");
+ Assert.assertEquals(message.getString(WorkflowExecutionArgs.TIMESTAMP.getName()),
+ "2012-01-01T01:00Z");
+ Assert.assertEquals(message.getString(WorkflowExecutionArgs.STATUS.getName()), "SUCCEEDED");
+ }
+}
[7/8] git commit: FALCON-492 - Refactor workflow listener
implementation init config driven. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-492 - Refactor workflow listener implementation init config driven. 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/1ccae452
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/1ccae452
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/1ccae452
Branch: refs/heads/master
Commit: 1ccae452933d75f212d3d267a3cf1312d705a4cc
Parents: a099443
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Fri Aug 8 13:02:53 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Fri Aug 8 13:02:53 2014 -0700
----------------------------------------------------------------------
.../WorkflowJobEndNotificationService.java | 17 +-
common/src/main/resources/startup.properties | 11 ++
.../WorkflowJobEndNotificationServiceTest.java | 163 +++++++++++++++++++
.../service/ProcessSubscriberService.java | 4 +-
.../falcon/rerun/service/LateRunService.java | 4 +-
.../falcon/rerun/service/RetryService.java | 4 +-
src/conf/startup.properties | 15 ++
7 files changed, 209 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/1ccae452/common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java b/common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java
index 0602214..39237f9 100644
--- a/common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java
+++ b/common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java
@@ -24,6 +24,8 @@ import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.resource.InstancesResult;
import org.apache.falcon.security.CurrentUser;
import org.apache.falcon.service.FalconService;
+import org.apache.falcon.util.ReflectionUtils;
+import org.apache.falcon.util.StartupProperties;
import org.apache.falcon.workflow.engine.AbstractWorkflowEngine;
import java.util.Date;
@@ -35,18 +37,27 @@ import java.util.Set;
*/
public class WorkflowJobEndNotificationService implements FalconService {
- public static final String NAME = WorkflowJobEndNotificationService.class.getSimpleName();
+ public static final String SERVICE_NAME = WorkflowJobEndNotificationService.class.getSimpleName();
private Set<WorkflowExecutionListener> listeners = new LinkedHashSet<WorkflowExecutionListener>();
@Override
public String getName() {
- return NAME;
+ return SERVICE_NAME;
}
@Override
public void init() throws FalconException {
- // do nothing
+ String listenerClassNames = StartupProperties.get().getProperty(
+ "workflow.execution.listeners");
+ for (String listenerClassName : listenerClassNames.split(",")) {
+ listenerClassName = listenerClassName.trim();
+ if (listenerClassName.isEmpty()) {
+ continue;
+ }
+ WorkflowExecutionListener listener = ReflectionUtils.getInstanceByClassName(listenerClassName);
+ registerListener(listener);
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/1ccae452/common/src/main/resources/startup.properties
----------------------------------------------------------------------
diff --git a/common/src/main/resources/startup.properties b/common/src/main/resources/startup.properties
index d215a62..314146c 100644
--- a/common/src/main/resources/startup.properties
+++ b/common/src/main/resources/startup.properties
@@ -20,6 +20,7 @@
######### Implementation classes #########
## DONT MODIFY UNLESS SURE ABOUT CHANGE ##
+
*.workflow.engine.impl=org.apache.falcon.workflow.engine.OozieWorkflowEngine
*.oozie.process.workflow.builder=org.apache.falcon.workflow.OozieProcessWorkflowBuilder
*.oozie.feed.workflow.builder=org.apache.falcon.workflow.OozieFeedWorkflowBuilder
@@ -28,6 +29,7 @@
*.ProcessInstanceManager.impl=org.apache.falcon.resource.InstanceManager
*.catalog.service.impl=org.apache.falcon.catalog.HiveCatalogService
+##### Falcon Services #####
*.application.services=org.apache.falcon.security.AuthenticationInitializationService,\
org.apache.falcon.workflow.WorkflowJobEndNotificationService, \
org.apache.falcon.service.ProcessSubscriberService,\
@@ -35,13 +37,22 @@
org.apache.falcon.rerun.service.RetryService,\
org.apache.falcon.rerun.service.LateRunService,\
org.apache.falcon.service.LogCleanupService
+
+##### Falcon Configuration Store Change listeners #####
*.configstore.listeners=org.apache.falcon.entity.v0.EntityGraph,\
org.apache.falcon.entity.ColoClusterRelation,\
org.apache.falcon.group.FeedGroupMap,\
org.apache.falcon.service.SharedLibraryHostingService
+
+##### JMS MQ Broker Implementation class #####
*.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
+##### Workflow Job Execution Completion listeners #####
+*.workflow.execution.listeners=
+
######### Implementation classes #########
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/1ccae452/common/src/test/java/org/apache/falcon/workflow/WorkflowJobEndNotificationServiceTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/workflow/WorkflowJobEndNotificationServiceTest.java b/common/src/test/java/org/apache/falcon/workflow/WorkflowJobEndNotificationServiceTest.java
new file mode 100644
index 0000000..2fe08e4
--- /dev/null
+++ b/common/src/test/java/org/apache/falcon/workflow/WorkflowJobEndNotificationServiceTest.java
@@ -0,0 +1,163 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * 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
+ *
+ * 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.
+ */
+
+package org.apache.falcon.workflow;
+
+import org.apache.falcon.FalconException;
+import org.apache.falcon.entity.v0.process.EngineType;
+import org.apache.falcon.util.StartupProperties;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * A test for WorkflowJobEndNotificationService.
+ */
+public class WorkflowJobEndNotificationServiceTest implements WorkflowExecutionListener {
+
+ private static final String FALCON_USER = "falcon-user";
+ private static final String LOGS_DIR = "target/log";
+ private static final String NOMINAL_TIME = "2014-01-01-01-00";
+ private static final String OPERATION = "GENERATE";
+
+ private static final String CLUSTER_NAME = "primary-cluster";
+ private static final String ENTITY_NAME = "sample-process";
+ private static final String WORKFLOW_NAME = "imp-click-join-workflow";
+ private static final String WORKFLOW_VERSION = "1.0.9";
+
+ private static final String INPUT_FEED_NAMES = "impression-feed#clicks-feed";
+ private static final String INPUT_INSTANCE_PATHS =
+ "jail://global:00/falcon/impression-feed/2014/01/01,jail://global:00/falcon/impression-feed/2014/01/02"
+ + "#jail://global:00/falcon/clicks-feed/2014-01-01";
+
+ private static final String OUTPUT_FEED_NAMES = "imp-click-join1,imp-click-join2";
+ private static final String OUTPUT_INSTANCE_PATHS =
+ "jail://global:00/falcon/imp-click-join1/20140101,jail://global:00/falcon/imp-click-join2/20140101";
+
+ private static final String BROKER = "org.apache.activemq.ActiveMQConnectionFactory";
+
+ private WorkflowJobEndNotificationService service;
+ private WorkflowExecutionContext savedContext;
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ service = new WorkflowJobEndNotificationService();
+ savedContext = WorkflowExecutionContext.create(getTestMessageArgs(),
+ WorkflowExecutionContext.Type.POST_PROCESSING);
+ Assert.assertNotNull(savedContext);
+ }
+
+ @AfterMethod
+ public void tearDown() throws Exception {
+ service.destroy();
+ }
+
+ @Test
+ public void testGetName() throws Exception {
+ Assert.assertEquals(service.getName(), WorkflowJobEndNotificationService.SERVICE_NAME);
+ }
+
+ @Test
+ public void testInit() throws Exception {
+ String listenerClassNames = StartupProperties.get().getProperty(
+ "workflow.execution.listeners");
+ Assert.assertEquals(listenerClassNames, "");
+
+ try {
+ StartupProperties.get().setProperty("workflow.execution.listeners",
+ "org.apache.falcon.workflow.WorkflowJobEndNotificationServiceTest");
+ listenerClassNames = StartupProperties.get().getProperty(
+ "workflow.execution.listeners");
+ Assert.assertEquals(listenerClassNames,
+ "org.apache.falcon.workflow.WorkflowJobEndNotificationServiceTest");
+
+ service.init();
+ notifyFailure(savedContext);
+ notifySuccess(savedContext);
+ } finally {
+ service.unregisterListener(this);
+ StartupProperties.get().setProperty("workflow.execution.listeners", "");
+ }
+ }
+
+ @Override
+ public void onSuccess(WorkflowExecutionContext context) throws FalconException {
+ Assert.assertNotNull(context);
+ Assert.assertEquals(context.entrySet().size(), 28);
+ }
+
+ @Override
+ public void onFailure(WorkflowExecutionContext context) throws FalconException {
+ Assert.assertNotNull(context);
+ Assert.assertEquals(context.entrySet().size(), 28);
+ }
+
+ private void notifyFailure(WorkflowExecutionContext context) {
+ try {
+ service.notifyFailure(context);
+ } catch (FalconException ignored) {
+ // do nothing
+ }
+ }
+
+ private void notifySuccess(WorkflowExecutionContext context) {
+ try {
+ service.notifySuccess(context);
+ } catch (FalconException ignored) {
+ // do nothing
+ }
+ }
+
+ private static String[] getTestMessageArgs() {
+ return new String[]{
+ "-" + WorkflowExecutionArgs.CLUSTER_NAME.getName(), CLUSTER_NAME,
+ "-" + WorkflowExecutionArgs.ENTITY_TYPE.getName(), "process",
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), ENTITY_NAME,
+ "-" + WorkflowExecutionArgs.NOMINAL_TIME.getName(), NOMINAL_TIME,
+ "-" + WorkflowExecutionArgs.OPERATION.getName(), OPERATION,
+
+ "-" + WorkflowExecutionArgs.INPUT_FEED_NAMES.getName(), INPUT_FEED_NAMES,
+ "-" + WorkflowExecutionArgs.INPUT_FEED_PATHS.getName(), INPUT_INSTANCE_PATHS,
+
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), OUTPUT_FEED_NAMES,
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(), OUTPUT_INSTANCE_PATHS,
+
+ "-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
+ "-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), FALCON_USER,
+ "-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
+ "-" + WorkflowExecutionArgs.STATUS.getName(), "SUCCEEDED",
+ "-" + WorkflowExecutionArgs.TIMESTAMP.getName(), NOMINAL_TIME,
+
+ "-" + WorkflowExecutionArgs.WF_ENGINE_URL.getName(), "http://localhost:11000/oozie",
+ "-" + WorkflowExecutionArgs.USER_SUBFLOW_ID.getName(), "userflow@wf-id",
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_NAME.getName(), WORKFLOW_NAME,
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_VERSION.getName(), WORKFLOW_VERSION,
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_ENGINE.getName(), EngineType.PIG.name(),
+
+ "-" + WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), BROKER,
+ "-" + WorkflowExecutionArgs.BRKR_URL.getName(), "tcp://localhost:61616?daemon=true",
+ "-" + WorkflowExecutionArgs.USER_BRKR_IMPL_CLASS.getName(), BROKER,
+ "-" + WorkflowExecutionArgs.USER_BRKR_URL.getName(), "tcp://localhost:61616?daemon=true",
+ "-" + WorkflowExecutionArgs.BRKR_TTL.getName(), "1000",
+
+ "-" + WorkflowExecutionArgs.LOG_DIR.getName(), LOGS_DIR,
+ "-" + WorkflowExecutionArgs.LOG_FILE.getName(), LOGS_DIR + "/log.txt",
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/1ccae452/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java b/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java
index 06fec9f..d17469b 100644
--- a/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java
+++ b/prism/src/main/java/org/apache/falcon/service/ProcessSubscriberService.java
@@ -50,7 +50,7 @@ public class ProcessSubscriberService implements FalconService {
@Override
public void init() throws FalconException {
- if (!Services.get().isRegistered(WorkflowJobEndNotificationService.NAME)) {
+ if (!Services.get().isRegistered(WorkflowJobEndNotificationService.SERVICE_NAME)) {
throw new FalconException("WorkflowJobEndNotificationService must be configured ahead");
}
@@ -59,7 +59,7 @@ public class ProcessSubscriberService implements FalconService {
String falconEntityTopic = getPropertyValue(JMSProps.FalconEntityTopic);
WorkflowJobEndNotificationService jobEndNotificationService =
- Services.get().getService(WorkflowJobEndNotificationService.NAME);
+ Services.get().getService(WorkflowJobEndNotificationService.SERVICE_NAME);
subscriber = new JMSMessageConsumer(falconBrokerImplClass, "", "", falconBrokerUrl,
falconEntityTopic, jobEndNotificationService);
subscriber.startSubscriber();
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/1ccae452/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java
----------------------------------------------------------------------
diff --git a/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java b/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java
index 01563b4..bbfdaff 100644
--- a/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java
+++ b/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java
@@ -44,7 +44,7 @@ public class LateRunService implements FalconService {
@Override
public void init() throws FalconException {
- if (!Services.get().isRegistered(WorkflowJobEndNotificationService.NAME)) {
+ if (!Services.get().isRegistered(WorkflowJobEndNotificationService.SERVICE_NAME)) {
throw new FalconException("WorkflowJobEndNotificationService must be configured ahead");
}
@@ -57,7 +57,7 @@ public class LateRunService implements FalconService {
rerunHandler.init(queue);
Services.get().<WorkflowJobEndNotificationService>getService(
- WorkflowJobEndNotificationService.NAME).registerListener(rerunHandler);
+ WorkflowJobEndNotificationService.SERVICE_NAME).registerListener(rerunHandler);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/1ccae452/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java
----------------------------------------------------------------------
diff --git a/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java b/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java
index 77bc9d5..771e2ce 100644
--- a/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java
+++ b/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java
@@ -47,7 +47,7 @@ public class RetryService implements FalconService {
@Override
public void init() throws FalconException {
- if (!Services.get().isRegistered(WorkflowJobEndNotificationService.NAME)) {
+ if (!Services.get().isRegistered(WorkflowJobEndNotificationService.SERVICE_NAME)) {
throw new FalconException("WorkflowJobEndNotificationService must be configured ahead");
}
@@ -57,7 +57,7 @@ public class RetryService implements FalconService {
rerunHandler.init(queue);
Services.get().<WorkflowJobEndNotificationService>getService(
- WorkflowJobEndNotificationService.NAME).registerListener(rerunHandler);
+ WorkflowJobEndNotificationService.SERVICE_NAME).registerListener(rerunHandler);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/1ccae452/src/conf/startup.properties
----------------------------------------------------------------------
diff --git a/src/conf/startup.properties b/src/conf/startup.properties
index 563c4b9..78466af 100644
--- a/src/conf/startup.properties
+++ b/src/conf/startup.properties
@@ -24,6 +24,7 @@
######### Implementation classes #########
## DONT MODIFY UNLESS SURE ABOUT CHANGE ##
+
*.workflow.engine.impl=org.apache.falcon.workflow.engine.OozieWorkflowEngine
*.oozie.process.workflow.builder=org.apache.falcon.workflow.OozieProcessWorkflowBuilder
*.oozie.feed.workflow.builder=org.apache.falcon.workflow.OozieFeedWorkflowBuilder
@@ -32,6 +33,7 @@
*.ProcessInstanceManager.impl=org.apache.falcon.resource.InstanceManager
*.catalog.service.impl=org.apache.falcon.catalog.HiveCatalogService
+##### Falcon Services #####
*.application.services=org.apache.falcon.security.AuthenticationInitializationService,\
org.apache.falcon.workflow.WorkflowJobEndNotificationService, \
org.apache.falcon.service.ProcessSubscriberService,\
@@ -40,17 +42,30 @@
org.apache.falcon.rerun.service.LateRunService,\
org.apache.falcon.metadata.MetadataMappingService,\
org.apache.falcon.service.LogCleanupService
+
+##### Prism Services #####
prism.application.services=org.apache.falcon.entity.store.ConfigurationStore
+
+##### Falcon Configuration Store Change listeners #####
*.configstore.listeners=org.apache.falcon.entity.v0.EntityGraph,\
org.apache.falcon.entity.ColoClusterRelation,\
org.apache.falcon.group.FeedGroupMap,\
org.apache.falcon.service.SharedLibraryHostingService
+
+##### Prism Configuration Store Change listeners #####
prism.configstore.listeners=org.apache.falcon.entity.v0.EntityGraph,\
org.apache.falcon.entity.ColoClusterRelation,\
org.apache.falcon.group.FeedGroupMap
+
+##### JMS MQ Broker Implementation class #####
*.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
+##### Workflow Job Execution Completion listeners #####
+*.workflow.execution.listeners=
+
######### Implementation classes #########
[8/8] git commit: FALCON-327 - Simplify message passing framework.
Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-327 - Simplify message passing framework. 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/9921457c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/9921457c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/9921457c
Branch: refs/heads/master
Commit: 9921457c36a606b19c74f2674ff556f8e410d429
Parents: 1ccae45
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Fri Aug 8 13:31:19 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Fri Aug 8 13:31:19 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 8 ++++++++
.../org/apache/falcon/metadata/MetadataMappingService.java | 4 ++--
.../resource/metadata/LineageMetadataResourceTest.java | 2 +-
3 files changed, 11 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/9921457c/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 7cc4c67..d26b224 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -19,6 +19,14 @@ Trunk (Unreleased)
FALCON-263 API to get workflow parameters. (pavan kumar kolamuri via Shwetha GS)
IMPROVEMENTS
+ FALCON-327 - Simplify message passing framework (Venkatesh Seetharam)
+ FALCON-484 - Refactor workflow related code into context, listener and Notification
+ FALCON-485 - Simplify JMS Message Sender/Consumer and use Workflow Context
+ FALCON-486 - Introduce Workflow Context in Post Processing
+ FALCON-487 -Introduce Workflow Context in Rerun framework
+ FALCON-488 - Introduce Workflow Context in Lineage Service
+ FALCON-492 - Refactor workflow listener implementation init config driven
+
FALCON-554 Extend time functions available in Falcon to support current &
last week (Contributed by Srikanth Sundarrajan)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/9921457c/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java b/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
index 80e96ba..a501e69 100644
--- a/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
+++ b/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
@@ -98,7 +98,7 @@ public class MetadataMappingService
ConfigurationStore.get().registerListener(this);
Services.get().<WorkflowJobEndNotificationService>getService(
- WorkflowJobEndNotificationService.NAME).registerListener(this);
+ WorkflowJobEndNotificationService.SERVICE_NAME).registerListener(this);
}
protected Graph initializeGraphDB() {
@@ -187,7 +187,7 @@ public class MetadataMappingService
@Override
public void destroy() throws FalconException {
Services.get().<WorkflowJobEndNotificationService>getService(
- WorkflowJobEndNotificationService.NAME).unregisterListener(this);
+ WorkflowJobEndNotificationService.SERVICE_NAME).unregisterListener(this);
LOG.info("Shutting down graph db");
graph.shutdown();
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/9921457c/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 a82acda..d9e9f86 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
@@ -97,7 +97,7 @@ public class LineageMetadataResourceTest {
configStore = ConfigurationStore.get();
Services.get().register(new WorkflowJobEndNotificationService());
- Assert.assertTrue(Services.get().isRegistered(WorkflowJobEndNotificationService.NAME));
+ Assert.assertTrue(Services.get().isRegistered(WorkflowJobEndNotificationService.SERVICE_NAME));
StartupProperties.get().setProperty("falcon.graph.preserve.history", "true");
service = new MetadataMappingService();
[6/8] git commit: FALCON-488 - Introduce Workflow Context in Lineage
Service. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-488 - Introduce Workflow Context in Lineage Service. 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/a0994437
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/a0994437
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/a0994437
Branch: refs/heads/master
Commit: a099443758045d625da5d2bb382898e5e13009ab
Parents: b9781d1
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Fri Aug 8 13:01:37 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Fri Aug 8 13:01:37 2014 -0700
----------------------------------------------------------------------
.../EntityRelationshipGraphBuilder.java | 6 +-
.../InstanceRelationshipGraphBuilder.java | 100 ++++++------
.../org/apache/falcon/metadata/LineageArgs.java | 82 ----------
.../apache/falcon/metadata/LineageRecorder.java | 154 -------------------
.../falcon/metadata/MetadataMappingService.java | 59 +++----
.../metadata/RelationshipGraphBuilder.java | 10 --
.../falcon/metadata/LineageRecorderTest.java | 103 -------------
.../metadata/MetadataMappingServiceTest.java | 71 +++++----
.../metadata/LineageMetadataResourceTest.java | 76 +++++----
9 files changed, 174 insertions(+), 487 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a0994437/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java b/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
index 2f46ff4..364d8f1 100644
--- a/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
+++ b/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
@@ -29,6 +29,7 @@ import org.apache.falcon.entity.v0.process.Output;
import org.apache.falcon.entity.v0.process.Outputs;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.entity.v0.process.Workflow;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -166,10 +167,11 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
}
public void addWorkflowProperties(Workflow workflow, Vertex processVertex, String processName) {
- processVertex.setProperty(LineageArgs.USER_WORKFLOW_NAME.getOptionName(),
+ processVertex.setProperty(WorkflowExecutionArgs.USER_WORKFLOW_NAME.getName(),
ProcessHelper.getProcessWorkflowName(workflow.getName(), processName));
processVertex.setProperty(RelationshipProperty.VERSION.getName(), workflow.getVersion());
- processVertex.setProperty(LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(), workflow.getEngine().value());
+ processVertex.setProperty(WorkflowExecutionArgs.USER_WORKFLOW_ENGINE.getName(),
+ workflow.getEngine().value());
}
public void updateWorkflowProperties(Workflow oldWorkflow, Workflow newWorkflow,
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a0994437/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java b/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
index 7ee185f..34857a3 100644
--- a/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
+++ b/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
@@ -34,9 +34,10 @@ import org.apache.falcon.entity.v0.feed.LocationType;
import org.apache.falcon.entity.v0.process.Process;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
import java.net.URISyntaxException;
-import java.util.Map;
/**
* Instance Metadata relationship mapping helper.
@@ -45,18 +46,17 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
private static final Logger LOG = LoggerFactory.getLogger(InstanceRelationshipGraphBuilder.class);
- private static final String PROCESS_INSTANCE_FORMAT = "yyyy-MM-dd-HH-mm"; // nominal time
private static final String FEED_INSTANCE_FORMAT = "yyyyMMddHHmm"; // computed
// process workflow properties from message
- private static final String[] INSTANCE_WORKFLOW_PROPERTIES = {
- LineageArgs.USER_WORKFLOW_NAME.getOptionName(),
- LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(),
- LineageArgs.WORKFLOW_ID.getOptionName(),
- LineageArgs.RUN_ID.getOptionName(),
- LineageArgs.STATUS.getOptionName(),
- LineageArgs.WF_ENGINE_URL.getOptionName(),
- LineageArgs.USER_SUBFLOW_ID.getOptionName(),
+ private static final WorkflowExecutionArgs[] INSTANCE_WORKFLOW_PROPERTIES = {
+ WorkflowExecutionArgs.USER_WORKFLOW_NAME,
+ WorkflowExecutionArgs.USER_WORKFLOW_ENGINE,
+ WorkflowExecutionArgs.WORKFLOW_ID,
+ WorkflowExecutionArgs.RUN_ID,
+ WorkflowExecutionArgs.STATUS,
+ WorkflowExecutionArgs.WF_ENGINE_URL,
+ WorkflowExecutionArgs.USER_SUBFLOW_ID,
};
@@ -64,49 +64,51 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
super(graph, preserveHistory);
}
- public Vertex addProcessInstance(Map<String, String> lineageMetadata) throws FalconException {
- String entityName = lineageMetadata.get(LineageArgs.ENTITY_NAME.getOptionName());
- String processInstanceName = getProcessInstanceName(entityName,
- lineageMetadata.get(LineageArgs.NOMINAL_TIME.getOptionName()));
- LOG.info("Adding process instance: {}", processInstanceName);
+ public Vertex addProcessInstance(WorkflowExecutionContext context) throws FalconException {
+ String processInstanceName = getProcessInstanceName(context);
+ LOG.info("Adding process instance: " + processInstanceName);
- String timestamp = getTimestamp(lineageMetadata);
- Vertex processInstance = addVertex(processInstanceName, RelationshipType.PROCESS_INSTANCE, timestamp);
- addWorkflowInstanceProperties(processInstance, lineageMetadata);
+ Vertex processInstance = addVertex(processInstanceName,
+ RelationshipType.PROCESS_INSTANCE, context.getTimeStampAsISO8601());
+ addWorkflowInstanceProperties(processInstance, context);
- addInstanceToEntity(processInstance, entityName,
+ addInstanceToEntity(processInstance, context.getEntityName(),
RelationshipType.PROCESS_ENTITY, RelationshipLabel.INSTANCE_ENTITY_EDGE);
- addInstanceToEntity(processInstance, lineageMetadata.get(LineageArgs.CLUSTER.getOptionName()),
+ addInstanceToEntity(processInstance, context.getClusterName(),
RelationshipType.CLUSTER_ENTITY, RelationshipLabel.PROCESS_CLUSTER_EDGE);
- addInstanceToEntity(processInstance, lineageMetadata.get(LineageArgs.WORKFLOW_USER.getOptionName()),
+ addInstanceToEntity(processInstance, context.getWorkflowUser(),
RelationshipType.USER, RelationshipLabel.USER);
if (isPreserveHistory()) {
- Process process = ConfigurationStore.get().get(EntityType.PROCESS, entityName);
+ Process process = ConfigurationStore.get().get(EntityType.PROCESS, context.getEntityName());
addDataClassification(process.getTags(), processInstance);
}
return processInstance;
}
- private String getTimestamp(Map<String, String> lineageMetadata) {
- String timestamp = lineageMetadata.get(LineageArgs.TIMESTAMP.getOptionName());
- return SchemaHelper.formatDateUTCToISO8601(timestamp, PROCESS_INSTANCE_FORMAT);
+ public String getProcessInstanceName(WorkflowExecutionContext context) {
+ return context.getEntityName() + "/" + context.getNominalTimeAsISO8601();
}
public void addWorkflowInstanceProperties(Vertex processInstance,
- Map<String, String> lineageMetadata) {
- for (String instanceWorkflowProperty : INSTANCE_WORKFLOW_PROPERTIES) {
- addProperty(processInstance, lineageMetadata, instanceWorkflowProperty);
+ WorkflowExecutionContext context) {
+ for (WorkflowExecutionArgs instanceWorkflowProperty : INSTANCE_WORKFLOW_PROPERTIES) {
+ addProperty(processInstance, context, instanceWorkflowProperty);
}
+
processInstance.setProperty(RelationshipProperty.VERSION.getName(),
- lineageMetadata.get(LineageArgs.USER_WORKFLOW_VERSION.getOptionName()));
+ context.getUserWorkflowVersion());
}
- public String getProcessInstanceName(String entityName,
- String nominalTime) {
- return entityName + "/"
- + SchemaHelper.formatDateUTCToISO8601(nominalTime, PROCESS_INSTANCE_FORMAT);
+ private void addProperty(Vertex vertex, WorkflowExecutionContext context,
+ WorkflowExecutionArgs optionName) {
+ String value = context.getValue(optionName);
+ if (value == null || value.length() == 0) {
+ return;
+ }
+
+ vertex.setProperty(optionName.getName(), value);
}
public void addInstanceToEntity(Vertex instanceVertex, String entityName,
@@ -122,37 +124,33 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
addEdge(instanceVertex, entityVertex, edgeLabel.getName());
}
- public void addOutputFeedInstances(Map<String, String> lineageMetadata,
+ public void addOutputFeedInstances(WorkflowExecutionContext context,
Vertex processInstance) throws FalconException {
- String outputFeedNamesArg = lineageMetadata.get(LineageArgs.FEED_NAMES.getOptionName());
+ String outputFeedNamesArg = context.getOutputFeedNames();
if ("NONE".equals(outputFeedNamesArg)) {
return; // there are no output feeds for this process
}
- String[] outputFeedNames = outputFeedNamesArg.split(",");
- String[] outputFeedInstancePaths =
- lineageMetadata.get(LineageArgs.FEED_INSTANCE_PATHS.getOptionName()).split(",");
+ String[] outputFeedNames = context.getOutputFeedNamesList();
+ String[] outputFeedInstancePaths = context.getOutputFeedInstancePathsList();
for (int index = 0; index < outputFeedNames.length; index++) {
String feedName = outputFeedNames[index];
String feedInstanceDataPath = outputFeedInstancePaths[index];
addFeedInstance(processInstance, RelationshipLabel.PROCESS_FEED_EDGE,
- lineageMetadata, feedName, feedInstanceDataPath);
+ context, feedName, feedInstanceDataPath);
}
}
- public void addInputFeedInstances(Map<String, String> lineageMetadata,
+ public void addInputFeedInstances(WorkflowExecutionContext context,
Vertex processInstance) throws FalconException {
- String inputFeedNamesArg = lineageMetadata.get(LineageArgs.INPUT_FEED_NAMES.getOptionName());
+ String inputFeedNamesArg = context.getInputFeedNames();
if ("NONE".equals(inputFeedNamesArg)) {
return; // there are no input feeds for this process
}
- String[] inputFeedNames =
- lineageMetadata.get(LineageArgs.INPUT_FEED_NAMES.getOptionName()).split("#");
- // Each input feed is separated by #
- String[] inputFeedInstancePaths =
- lineageMetadata.get(LineageArgs.INPUT_FEED_PATHS.getOptionName()).split("#");
+ String[] inputFeedNames = context.getInputFeedNamesList();
+ String[] inputFeedInstancePaths = context.getInputFeedInstancePathsList();
for (int index = 0; index < inputFeedNames.length; index++) {
String inputFeedName = inputFeedNames[index];
@@ -162,21 +160,21 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
for (String feedInstanceDataPath : feedInstancePaths) {
addFeedInstance(processInstance, RelationshipLabel.FEED_PROCESS_EDGE,
- lineageMetadata, inputFeedName, feedInstanceDataPath);
+ context, inputFeedName, feedInstanceDataPath);
}
}
}
private void addFeedInstance(Vertex processInstance, RelationshipLabel edgeLabel,
- Map<String, String> lineageMetadata, String feedName,
+ WorkflowExecutionContext context, String feedName,
String feedInstanceDataPath) throws FalconException {
- String clusterName = lineageMetadata.get(LineageArgs.CLUSTER.getOptionName());
+ String clusterName = context.getClusterName();
LOG.info("Computing feed instance for : name=" + feedName + ", path= "
+ feedInstanceDataPath + ", in cluster: " + clusterName);
String feedInstanceName = getFeedInstanceName(feedName, clusterName, feedInstanceDataPath);
LOG.info("Adding feed instance: " + feedInstanceName);
Vertex feedInstance = addVertex(feedInstanceName, RelationshipType.FEED_INSTANCE,
- getTimestamp(lineageMetadata));
+ context.getTimeStampAsISO8601());
addProcessFeedEdge(processInstance, feedInstance, edgeLabel);
@@ -184,7 +182,7 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
RelationshipType.FEED_ENTITY, RelationshipLabel.INSTANCE_ENTITY_EDGE);
addInstanceToEntity(feedInstance, clusterName,
RelationshipType.CLUSTER_ENTITY, RelationshipLabel.FEED_CLUSTER_EDGE);
- addInstanceToEntity(feedInstance, lineageMetadata.get(LineageArgs.WORKFLOW_USER.getOptionName()),
+ addInstanceToEntity(feedInstance, context.getWorkflowUser(),
RelationshipType.USER, RelationshipLabel.USER);
if (isPreserveHistory()) {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a0994437/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java b/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
deleted file mode 100644
index e91f7ab..0000000
--- a/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-
-package org.apache.falcon.metadata;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.Option;
-
-/**
- * Args for data Lineage.
- */
-@Deprecated // delete this class
-public enum LineageArgs {
- // process instance
- NOMINAL_TIME("nominalTime", "instance time"),
- ENTITY_TYPE("entityType", "type of the entity"),
- ENTITY_NAME("entityName", "name of the entity"),
- TIMESTAMP("timeStamp", "current timestamp"),
-
- // where
- CLUSTER("cluster", "name of the current cluster"),
- OPERATION("operation", "operation like generate, delete, replicate"),
-
- // who
- WORKFLOW_USER("workflowUser", "user who owns the feed instance (partition)"),
-
- // workflow details
- WORKFLOW_ID("workflowId", "current workflow-id of the instance"),
- RUN_ID("runId", "current run-id of the instance"),
- STATUS("status", "status of the user workflow isnstance"),
- WF_ENGINE_URL("workflowEngineUrl", "url of workflow engine server, ex:oozie"),
- USER_SUBFLOW_ID("subflowId", "external id of user workflow"),
- USER_WORKFLOW_ENGINE("userWorkflowEngine", "user workflow engine type"),
- USER_WORKFLOW_NAME("userWorkflowName", "user workflow name"),
- USER_WORKFLOW_VERSION("userWorkflowVersion", "user workflow version"),
-
- // what inputs
- INPUT_FEED_NAMES("falconInputFeeds", "name of the feeds which are used as inputs"),
- INPUT_FEED_PATHS("falconInputPaths", "comma separated input feed instance paths"),
-
- // what outputs
- FEED_NAMES("feedNames", "name of the feeds which are generated/replicated/deleted"),
- FEED_INSTANCE_PATHS("feedInstancePaths", "comma separated feed instance paths"),
-
- // lineage data recorded
- LOG_DIR("logDir", "log dir where lineage can be recorded");
-
- private String name;
- private String description;
-
- LineageArgs(String name, String description) {
- this.name = name;
- this.description = description;
- }
-
- public Option getOption() {
- return new Option(this.name, true, this.description);
- }
-
- public String getOptionName() {
- return this.name;
- }
-
- public String getOptionValue(CommandLine cmd) {
- return cmd.getOptionValue(this.name);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a0994437/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java b/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
deleted file mode 100644
index 5ab1bf0..0000000
--- a/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
+++ /dev/null
@@ -1,154 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-
-package org.apache.falcon.metadata;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.GnuParser;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.apache.falcon.FalconException;
-import org.apache.falcon.hadoop.HadoopClientFactory;
-import org.apache.hadoop.conf.Configured;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.permission.FsAction;
-import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.util.Tool;
-import org.apache.hadoop.util.ToolRunner;
-import org.json.simple.JSONValue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.OutputStream;
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * Utility called in the post process of oozie workflow to record lineage information.
- */
-@Deprecated // delete this class
-public class LineageRecorder extends Configured implements Tool {
-
- private static final Logger LOG = LoggerFactory.getLogger(LineageRecorder.class);
-
- public static void main(String[] args) throws Exception {
- ToolRunner.run(new LineageRecorder(), args);
- }
-
- @Override
- public int run(String[] arguments) throws Exception {
- CommandLine command = getCommand(arguments);
-
- LOG.info("Parsing lineage metadata from: {}", command);
- Map<String, String> lineageMetadata = getLineageMetadata(command);
- LOG.info("Lineage Metadata: {}", lineageMetadata);
-
- String lineageFile = getFilePath(command.getOptionValue(LineageArgs.LOG_DIR.getOptionName()),
- command.getOptionValue(LineageArgs.ENTITY_NAME.getOptionName())
- );
-
- LOG.info("Persisting lineage metadata to: {}", lineageFile);
- persistLineageMetadata(lineageMetadata, lineageFile);
-
- return 0;
- }
-
- protected static CommandLine getCommand(String[] arguments) throws ParseException {
-
- Options options = new Options();
-
- for (LineageArgs arg : LineageArgs.values()) {
- addOption(options, arg);
- }
-
- return new GnuParser().parse(options, arguments);
- }
-
- private static void addOption(Options options, LineageArgs arg) {
- addOption(options, arg, true);
- }
-
- private static void addOption(Options options, LineageArgs arg, boolean isRequired) {
- Option option = arg.getOption();
- option.setRequired(isRequired);
- options.addOption(option);
- }
-
- protected Map<String, String> getLineageMetadata(CommandLine command) {
- Map<String, String> lineageMetadata = new HashMap<String, String>();
-
- for (LineageArgs arg : LineageArgs.values()) {
- lineageMetadata.put(arg.getOptionName(), arg.getOptionValue(command));
- }
-
- return lineageMetadata;
- }
-
- public static String getFilePath(String logDir, String entityName) {
- return logDir + entityName + "-lineage.json";
- }
-
- /**
- * this method is invoked from with in the workflow.
- *
- * @param lineageMetadata metadata to persist
- * @param lineageFile file to serialize the metadata
- * @throws IOException
- * @throws FalconException
- */
- protected void persistLineageMetadata(Map<String, String> lineageMetadata,
- String lineageFile) throws IOException, FalconException {
- OutputStream out = null;
- Path file = new Path(lineageFile);
- try {
- FileSystem fs = HadoopClientFactory.get().createFileSystem(file.toUri(), getConf());
- out = fs.create(file);
-
- // making sure falcon can read this file
- FsPermission permission = new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL);
- fs.setPermission(file, permission);
-
- out.write(JSONValue.toJSONString(lineageMetadata).getBytes());
- } finally {
- if (out != null) {
- try {
- out.close();
- } catch (IOException ignore) {
- // ignore
- }
- }
- }
- }
-
- public static Map<String, String> parseLineageMetadata(String lineageFile) throws FalconException {
- try {
- Path lineageDataPath = new Path(lineageFile); // file has 777 permissions
- FileSystem fs = HadoopClientFactory.get().createFileSystem(lineageDataPath.toUri());
-
- BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(lineageDataPath)));
- return (Map<String, String>) JSONValue.parse(in);
- } catch (IOException e) {
- throw new FalconException("Error opening lineage file", e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a0994437/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java b/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
index 5df4611..80e96ba 100644
--- a/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
+++ b/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
@@ -36,9 +36,13 @@ import org.apache.falcon.entity.v0.feed.Feed;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.service.ConfigurationChangeListener;
import org.apache.falcon.service.FalconService;
+import org.apache.falcon.service.Services;
import org.apache.falcon.util.StartupProperties;
+import org.apache.falcon.workflow.WorkflowJobEndNotificationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
+import org.apache.falcon.workflow.WorkflowExecutionListener;
import java.util.Map;
import java.util.Properties;
@@ -47,7 +51,8 @@ import java.util.Set;
/**
* Metadata relationship mapping service. Maps relationships into a graph database.
*/
-public class MetadataMappingService implements FalconService, ConfigurationChangeListener {
+public class MetadataMappingService
+ implements FalconService, ConfigurationChangeListener, WorkflowExecutionListener {
private static final Logger LOG = LoggerFactory.getLogger(MetadataMappingService.class);
@@ -92,6 +97,8 @@ public class MetadataMappingService implements FalconService, ConfigurationChang
instanceGraphBuilder = new InstanceRelationshipGraphBuilder(graph, preserveHistory);
ConfigurationStore.get().registerListener(this);
+ Services.get().<WorkflowJobEndNotificationService>getService(
+ WorkflowJobEndNotificationService.NAME).registerListener(this);
}
protected Graph initializeGraphDB() {
@@ -179,6 +186,9 @@ public class MetadataMappingService implements FalconService, ConfigurationChang
@Override
public void destroy() throws FalconException {
+ Services.get().<WorkflowJobEndNotificationService>getService(
+ WorkflowJobEndNotificationService.NAME).unregisterListener(this);
+
LOG.info("Shutting down graph db");
graph.shutdown();
}
@@ -244,54 +254,47 @@ public class MetadataMappingService implements FalconService, ConfigurationChang
// are already added to the graph
}
- /**
- * Entity operations.
- */
- public enum EntityOperations {
- GENERATE, REPLICATE, DELETE
- }
-
- public void onSuccessfulWorkflowCompletion(String entityName, String operation,
- String logDir) throws FalconException {
- String lineageFile = LineageRecorder.getFilePath(logDir, entityName);
-
- LOG.info("Parsing lineage metadata from: {}", lineageFile);
- Map<String, String> lineageMetadata = LineageRecorder.parseLineageMetadata(lineageFile);
-
- EntityOperations entityOperation = EntityOperations.valueOf(operation);
+ @Override
+ public void onSuccess(WorkflowExecutionContext context) throws FalconException {
+ WorkflowExecutionContext.EntityOperations entityOperation = context.getOperation();
- LOG.info("Adding lineage for entity: {}, operation: {}", entityName, operation);
+ LOG.info("Adding lineage for context {}", context);
switch (entityOperation) {
case GENERATE:
- onProcessInstanceAdded(lineageMetadata);
+ onProcessInstanceExecuted(context);
getTransactionalGraph().commit();
break;
case REPLICATE:
- onFeedInstanceReplicated(lineageMetadata);
+ onFeedInstanceReplicated(context);
break;
case DELETE:
- onFeedInstanceEvicted(lineageMetadata);
+ onFeedInstanceEvicted(context);
break;
default:
}
}
- private void onProcessInstanceAdded(Map<String, String> lineageMetadata) throws FalconException {
- Vertex processInstance = instanceGraphBuilder.addProcessInstance(lineageMetadata);
- instanceGraphBuilder.addOutputFeedInstances(lineageMetadata, processInstance);
- instanceGraphBuilder.addInputFeedInstances(lineageMetadata, processInstance);
+ @Override
+ public void onFailure(WorkflowExecutionContext context) throws FalconException {
+ // do nothing since lineage is only recorded for successful workflow
+ }
+
+ private void onProcessInstanceExecuted(WorkflowExecutionContext context) throws FalconException {
+ Vertex processInstance = instanceGraphBuilder.addProcessInstance(context);
+ instanceGraphBuilder.addOutputFeedInstances(context, processInstance);
+ instanceGraphBuilder.addInputFeedInstances(context, processInstance);
}
- private void onFeedInstanceReplicated(Map<String, String> lineageMetadata) {
- LOG.info("Adding replicated feed instance: {}", lineageMetadata.get(LineageArgs.NOMINAL_TIME.getOptionName()));
+ private void onFeedInstanceReplicated(WorkflowExecutionContext context) {
+ LOG.info("Adding replicated feed instance: {}", context.getNominalTimeAsISO8601());
// todo - tbd
}
- private void onFeedInstanceEvicted(Map<String, String> lineageMetadata) {
- LOG.info("Adding evicted feed instance: {}", lineageMetadata.get(LineageArgs.NOMINAL_TIME.getOptionName()));
+ private void onFeedInstanceEvicted(WorkflowExecutionContext context) {
+ LOG.info("Adding evicted feed instance: {}", context.getNominalTimeAsISO8601());
// todo - tbd
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a0994437/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java b/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
index 9ee0ea6..640df41 100644
--- a/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
+++ b/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
@@ -30,7 +30,6 @@ import org.slf4j.LoggerFactory;
import java.util.Date;
import java.util.Iterator;
-import java.util.Map;
/**
* Base class for Metadata relationship mapping helper.
@@ -192,13 +191,4 @@ public abstract class RelationshipGraphBuilder {
protected String getCurrentTimeStamp() {
return SchemaHelper.formatDateUTC(new Date());
}
-
- protected void addProperty(Vertex vertex, Map<String, String> lineageMetadata, String optionName) {
- String value = lineageMetadata.get(optionName);
- if (value == null || value.length() == 0) {
- return;
- }
-
- vertex.setProperty(optionName, value);
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a0994437/common/src/test/java/org/apache/falcon/metadata/LineageRecorderTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/metadata/LineageRecorderTest.java b/common/src/test/java/org/apache/falcon/metadata/LineageRecorderTest.java
deleted file mode 100644
index d686ab6..0000000
--- a/common/src/test/java/org/apache/falcon/metadata/LineageRecorderTest.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * 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
- *
- * 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.
- */
-
-package org.apache.falcon.metadata;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.testng.Assert;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeMethod;
-import org.testng.annotations.Test;
-
-import java.util.Map;
-
-/**
- * Unit test for lineage recorder.
- */
-public class LineageRecorderTest {
-
- private static final String LOGS_DIR = "target/log";
- private static final String NOMINAL_TIME = "2014-01-01-01-00";
- private static final String ENTITY_NAME = "test-process-entity";
-
- private String[] args;
-
- @BeforeMethod
- public void setUp() throws Exception {
- args = new String[]{
- "-" + LineageArgs.ENTITY_NAME.getOptionName(), ENTITY_NAME,
- "-" + LineageArgs.FEED_NAMES.getOptionName(), "out-click-logs,out-raw-logs",
- "-" + LineageArgs.FEED_INSTANCE_PATHS.getOptionName(),
- "/out-click-logs/10/05/05/00/20,/out-raw-logs/10/05/05/00/20",
- "-" + LineageArgs.WORKFLOW_ID.getOptionName(), "workflow-01-00",
- "-" + LineageArgs.WORKFLOW_USER.getOptionName(), "falcon",
- "-" + LineageArgs.RUN_ID.getOptionName(), "1",
- "-" + LineageArgs.NOMINAL_TIME.getOptionName(), NOMINAL_TIME,
- "-" + LineageArgs.TIMESTAMP.getOptionName(), "2014-01-01-01-00",
- "-" + LineageArgs.ENTITY_TYPE.getOptionName(), ("process"),
- "-" + LineageArgs.OPERATION.getOptionName(), ("GENERATE"),
- "-" + LineageArgs.STATUS.getOptionName(), ("SUCCEEDED"),
- "-" + LineageArgs.CLUSTER.getOptionName(), "corp",
- "-" + LineageArgs.WF_ENGINE_URL.getOptionName(), "http://localhost:11000/oozie/",
- "-" + LineageArgs.LOG_DIR.getOptionName(), LOGS_DIR,
- "-" + LineageArgs.USER_SUBFLOW_ID.getOptionName(), "userflow@wf-id" + "test",
- "-" + LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(), "oozie",
- "-" + LineageArgs.INPUT_FEED_NAMES.getOptionName(), "in-click-logs,in-raw-logs",
- "-" + LineageArgs.INPUT_FEED_PATHS.getOptionName(),
- "/in-click-logs/10/05/05/00/20,/in-raw-logs/10/05/05/00/20",
- "-" + LineageArgs.USER_WORKFLOW_NAME.getOptionName(), "test-workflow",
- "-" + LineageArgs.USER_WORKFLOW_VERSION.getOptionName(), "1.0.0",
- };
- }
-
- @AfterMethod
- public void tearDown() throws Exception {
-
- }
-
- @Test
- public void testMain() throws Exception {
- LineageRecorder lineageRecorder = new LineageRecorder();
- CommandLine command = LineageRecorder.getCommand(args);
- Map<String, String> lineageMetadata = lineageRecorder.getLineageMetadata(command);
-
- LineageRecorder.main(args);
-
- String lineageFile = LineageRecorder.getFilePath(LOGS_DIR, ENTITY_NAME);
- Path lineageDataPath = new Path(lineageFile);
- FileSystem fs = lineageDataPath.getFileSystem(new Configuration());
- Assert.assertTrue(fs.exists(lineageDataPath));
-
- Map<String, String> recordedLineageMetadata =
- LineageRecorder.parseLineageMetadata(lineageFile);
-
- for (Map.Entry<String, String> entry : lineageMetadata.entrySet()) {
- Assert.assertEquals(lineageMetadata.get(entry.getKey()),
- recordedLineageMetadata.get(entry.getKey()));
- }
- }
-
- @Test
- public void testGetFilePath() throws Exception {
- String path = LOGS_DIR + ENTITY_NAME + "-lineage.json";
- Assert.assertEquals(path, LineageRecorder.getFilePath(LOGS_DIR, ENTITY_NAME));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a0994437/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java b/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
index 52bb7ae..c94de10 100644
--- a/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
+++ b/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
@@ -36,7 +36,11 @@ import org.apache.falcon.entity.v0.feed.Locations;
import org.apache.falcon.entity.v0.process.EngineType;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.security.CurrentUser;
+import org.apache.falcon.service.Services;
import org.apache.falcon.util.StartupProperties;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
+import org.apache.falcon.workflow.WorkflowJobEndNotificationService;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
@@ -75,6 +79,8 @@ public class MetadataMappingServiceTest {
public static final String OUTPUT_INSTANCE_PATHS =
"jail://global:00/falcon/imp-click-join1/20140101,jail://global:00/falcon/imp-click-join2/20140101";
+ public static final String BROKER = "org.apache.activemq.ActiveMQConnectionFactory";
+
private ConfigurationStore configStore;
private MetadataMappingService service;
@@ -91,6 +97,7 @@ public class MetadataMappingServiceTest {
configStore = ConfigurationStore.get();
+ Services.get().register(new WorkflowJobEndNotificationService());
StartupProperties.get().setProperty("falcon.graph.preserve.history", "true");
service = new MetadataMappingService();
service.init();
@@ -218,9 +225,9 @@ public class MetadataMappingServiceTest {
service.destroy();
service.init();
- LineageRecorder.main(getTestMessageArgs());
-
- service.onSuccessfulWorkflowCompletion(PROCESS_ENTITY_NAME, OPERATION, LOGS_DIR);
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(getTestMessageArgs(),
+ WorkflowExecutionContext.Type.POST_PROCESSING);
+ service.onSuccess(context);
debug(service.getGraph());
GraphUtils.dump(service.getGraph());
@@ -592,31 +599,39 @@ public class MetadataMappingServiceTest {
private static String[] getTestMessageArgs() {
return new String[]{
- "-" + LineageArgs.NOMINAL_TIME.getOptionName(), NOMINAL_TIME,
- "-" + LineageArgs.TIMESTAMP.getOptionName(), NOMINAL_TIME,
-
- "-" + LineageArgs.ENTITY_NAME.getOptionName(), PROCESS_ENTITY_NAME,
- "-" + LineageArgs.ENTITY_TYPE.getOptionName(), ("process"),
- "-" + LineageArgs.CLUSTER.getOptionName(), CLUSTER_ENTITY_NAME,
- "-" + LineageArgs.OPERATION.getOptionName(), OPERATION,
-
- "-" + LineageArgs.INPUT_FEED_NAMES.getOptionName(), INPUT_FEED_NAMES,
- "-" + LineageArgs.INPUT_FEED_PATHS.getOptionName(), INPUT_INSTANCE_PATHS,
-
- "-" + LineageArgs.FEED_NAMES.getOptionName(), OUTPUT_FEED_NAMES,
- "-" + LineageArgs.FEED_INSTANCE_PATHS.getOptionName(), OUTPUT_INSTANCE_PATHS,
-
- "-" + LineageArgs.WORKFLOW_ID.getOptionName(), "workflow-01-00",
- "-" + LineageArgs.WORKFLOW_USER.getOptionName(), FALCON_USER,
- "-" + LineageArgs.RUN_ID.getOptionName(), "1",
- "-" + LineageArgs.STATUS.getOptionName(), "SUCCEEDED",
- "-" + LineageArgs.WF_ENGINE_URL.getOptionName(), "http://localhost:11000/oozie",
- "-" + LineageArgs.USER_SUBFLOW_ID.getOptionName(), "userflow@wf-id",
- "-" + LineageArgs.USER_WORKFLOW_NAME.getOptionName(), WORKFLOW_NAME,
- "-" + LineageArgs.USER_WORKFLOW_VERSION.getOptionName(), WORKFLOW_VERSION,
- "-" + LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(), EngineType.PIG.name(),
-
- "-" + LineageArgs.LOG_DIR.getOptionName(), LOGS_DIR,
+ "-" + WorkflowExecutionArgs.CLUSTER_NAME.getName(), CLUSTER_ENTITY_NAME,
+ "-" + WorkflowExecutionArgs.ENTITY_TYPE.getName(), ("process"),
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), PROCESS_ENTITY_NAME,
+ "-" + WorkflowExecutionArgs.NOMINAL_TIME.getName(), NOMINAL_TIME,
+ "-" + WorkflowExecutionArgs.OPERATION.getName(), OPERATION,
+
+ "-" + WorkflowExecutionArgs.INPUT_FEED_NAMES.getName(), INPUT_FEED_NAMES,
+ "-" + WorkflowExecutionArgs.INPUT_FEED_PATHS.getName(), INPUT_INSTANCE_PATHS,
+
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), OUTPUT_FEED_NAMES,
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(), OUTPUT_INSTANCE_PATHS,
+
+ "-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
+ "-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), FALCON_USER,
+ "-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
+ "-" + WorkflowExecutionArgs.STATUS.getName(), "SUCCEEDED",
+ "-" + WorkflowExecutionArgs.TIMESTAMP.getName(), NOMINAL_TIME,
+
+ "-" + WorkflowExecutionArgs.WF_ENGINE_URL.getName(), "http://localhost:11000/oozie",
+ "-" + WorkflowExecutionArgs.USER_SUBFLOW_ID.getName(), "userflow@wf-id",
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_NAME.getName(), WORKFLOW_NAME,
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_VERSION.getName(), WORKFLOW_VERSION,
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_ENGINE.getName(), EngineType.PIG.name(),
+
+
+ "-" + WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), BROKER,
+ "-" + WorkflowExecutionArgs.BRKR_URL.getName(), "tcp://localhost:61616?daemon=true",
+ "-" + WorkflowExecutionArgs.USER_BRKR_IMPL_CLASS.getName(), BROKER,
+ "-" + WorkflowExecutionArgs.USER_BRKR_URL.getName(), "tcp://localhost:61616?daemon=true",
+ "-" + WorkflowExecutionArgs.BRKR_TTL.getName(), "1000",
+
+ "-" + WorkflowExecutionArgs.LOG_DIR.getName(), LOGS_DIR,
+ "-" + WorkflowExecutionArgs.LOG_FILE.getName(), LOGS_DIR + "/log.txt",
};
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/a0994437/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 d664782..a82acda 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
@@ -34,14 +34,15 @@ import org.apache.falcon.entity.v0.feed.LocationType;
import org.apache.falcon.entity.v0.feed.Locations;
import org.apache.falcon.entity.v0.process.EngineType;
import org.apache.falcon.entity.v0.process.Process;
-import org.apache.falcon.metadata.LineageArgs;
-import org.apache.falcon.metadata.LineageRecorder;
import org.apache.falcon.metadata.MetadataMappingService;
import org.apache.falcon.metadata.RelationshipLabel;
import org.apache.falcon.metadata.RelationshipProperty;
import org.apache.falcon.security.CurrentUser;
import org.apache.falcon.service.Services;
import org.apache.falcon.util.StartupProperties;
+import org.apache.falcon.workflow.WorkflowExecutionArgs;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
+import org.apache.falcon.workflow.WorkflowJobEndNotificationService;
import org.json.simple.JSONValue;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -95,10 +96,14 @@ public class LineageMetadataResourceTest {
configStore = ConfigurationStore.get();
+ Services.get().register(new WorkflowJobEndNotificationService());
+ Assert.assertTrue(Services.get().isRegistered(WorkflowJobEndNotificationService.NAME));
+
StartupProperties.get().setProperty("falcon.graph.preserve.history", "true");
service = new MetadataMappingService();
service.init();
Services.get().register(service);
+ Assert.assertTrue(Services.get().isRegistered(MetadataMappingService.SERVICE_NAME));
addClusterEntity();
addFeedEntity();
@@ -110,8 +115,11 @@ public class LineageMetadataResourceTest {
public void tearDown() throws Exception {
cleanupGraphStore(service.getGraph());
cleanupConfigurationStore(configStore);
+
service.destroy();
+
StartupProperties.get().setProperty("falcon.graph.preserve.history", "false");
+ Services.get().reset();
}
@Test (expectedExceptions = WebApplicationException.class)
@@ -397,6 +405,7 @@ public class LineageMetadataResourceTest {
Assert.assertEquals(response.getStatus(), Response.Status.NOT_FOUND.getStatusCode());
Assert.assertEquals(response.getEntity().toString(), "Lineage Metadata Service is not enabled.");
} finally {
+ Services.get().register(new WorkflowJobEndNotificationService());
Services.get().register(service);
}
}
@@ -517,37 +526,46 @@ public class LineageMetadataResourceTest {
}
public void addInstance() throws Exception {
- LineageRecorder.main(getTestMessageArgs());
- service.onSuccessfulWorkflowCompletion(PROCESS_ENTITY_NAME, OPERATION, LOGS_DIR);
+ WorkflowExecutionContext context = WorkflowExecutionContext.create(getTestMessageArgs(),
+ WorkflowExecutionContext.Type.POST_PROCESSING);
+ service.onSuccess(context);
}
private static String[] getTestMessageArgs() {
return new String[]{
- "-" + LineageArgs.NOMINAL_TIME.getOptionName(), NOMINAL_TIME,
- "-" + LineageArgs.TIMESTAMP.getOptionName(), NOMINAL_TIME,
-
- "-" + LineageArgs.ENTITY_NAME.getOptionName(), PROCESS_ENTITY_NAME,
- "-" + LineageArgs.ENTITY_TYPE.getOptionName(), ("process"),
- "-" + LineageArgs.CLUSTER.getOptionName(), CLUSTER_ENTITY_NAME,
- "-" + LineageArgs.OPERATION.getOptionName(), OPERATION,
-
- "-" + LineageArgs.INPUT_FEED_NAMES.getOptionName(), INPUT_FEED_NAMES,
- "-" + LineageArgs.INPUT_FEED_PATHS.getOptionName(), INPUT_INSTANCE_PATHS,
-
- "-" + LineageArgs.FEED_NAMES.getOptionName(), OUTPUT_FEED_NAMES,
- "-" + LineageArgs.FEED_INSTANCE_PATHS.getOptionName(), OUTPUT_INSTANCE_PATHS,
-
- "-" + LineageArgs.WORKFLOW_ID.getOptionName(), "workflow-01-00",
- "-" + LineageArgs.WORKFLOW_USER.getOptionName(), FALCON_USER,
- "-" + LineageArgs.RUN_ID.getOptionName(), "1",
- "-" + LineageArgs.STATUS.getOptionName(), "SUCCEEDED",
- "-" + LineageArgs.WF_ENGINE_URL.getOptionName(), "http://localhost:11000/oozie",
- "-" + LineageArgs.USER_SUBFLOW_ID.getOptionName(), "userflow@wf-id",
- "-" + LineageArgs.USER_WORKFLOW_NAME.getOptionName(), WORKFLOW_NAME,
- "-" + LineageArgs.USER_WORKFLOW_VERSION.getOptionName(), WORKFLOW_VERSION,
- "-" + LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(), EngineType.PIG.name(),
-
- "-" + LineageArgs.LOG_DIR.getOptionName(), LOGS_DIR,
+ "-" + WorkflowExecutionArgs.CLUSTER_NAME.getName(), CLUSTER_ENTITY_NAME,
+ "-" + WorkflowExecutionArgs.ENTITY_TYPE.getName(), ("process"),
+ "-" + WorkflowExecutionArgs.ENTITY_NAME.getName(), PROCESS_ENTITY_NAME,
+ "-" + WorkflowExecutionArgs.NOMINAL_TIME.getName(), NOMINAL_TIME,
+ "-" + WorkflowExecutionArgs.OPERATION.getName(), OPERATION,
+
+ "-" + WorkflowExecutionArgs.INPUT_FEED_NAMES.getName(), INPUT_FEED_NAMES,
+ "-" + WorkflowExecutionArgs.INPUT_FEED_PATHS.getName(), INPUT_INSTANCE_PATHS,
+
+ "-" + WorkflowExecutionArgs.FEED_NAMES.getName(), OUTPUT_FEED_NAMES,
+ "-" + WorkflowExecutionArgs.FEED_INSTANCE_PATHS.getName(), OUTPUT_INSTANCE_PATHS,
+
+ "-" + WorkflowExecutionArgs.WORKFLOW_ID.getName(), "workflow-01-00",
+ "-" + WorkflowExecutionArgs.WORKFLOW_USER.getName(), FALCON_USER,
+ "-" + WorkflowExecutionArgs.RUN_ID.getName(), "1",
+ "-" + WorkflowExecutionArgs.STATUS.getName(), "SUCCEEDED",
+ "-" + WorkflowExecutionArgs.TIMESTAMP.getName(), NOMINAL_TIME,
+
+ "-" + WorkflowExecutionArgs.WF_ENGINE_URL.getName(), "http://localhost:11000/oozie",
+ "-" + WorkflowExecutionArgs.USER_SUBFLOW_ID.getName(), "userflow@wf-id",
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_NAME.getName(), WORKFLOW_NAME,
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_VERSION.getName(), WORKFLOW_VERSION,
+ "-" + WorkflowExecutionArgs.USER_WORKFLOW_ENGINE.getName(), EngineType.PIG.name(),
+
+
+ "-" + WorkflowExecutionArgs.BRKR_IMPL_CLASS.getName(), "blah",
+ "-" + WorkflowExecutionArgs.BRKR_URL.getName(), "tcp://localhost:61616?daemon=true",
+ "-" + WorkflowExecutionArgs.USER_BRKR_IMPL_CLASS.getName(), "blah",
+ "-" + WorkflowExecutionArgs.USER_BRKR_URL.getName(), "tcp://localhost:61616?daemon=true",
+ "-" + WorkflowExecutionArgs.BRKR_TTL.getName(), "1000",
+
+ "-" + WorkflowExecutionArgs.LOG_DIR.getName(), LOGS_DIR,
+ "-" + WorkflowExecutionArgs.LOG_FILE.getName(), LOGS_DIR + "/log.txt",
};
}
[5/8] git commit: FALCON-487 -Introduce Workflow Context in Rerun
framework. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-487 -Introduce Workflow Context in Rerun framework. 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/b9781d12
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/b9781d12
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/b9781d12
Branch: refs/heads/master
Commit: b9781d121cc334a76c7f5dd0dde32ca3c257db91
Parents: c600036
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Fri Aug 8 12:59:38 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Fri Aug 8 12:59:38 2014 -0700
----------------------------------------------------------------------
.../rerun/handler/AbstractRerunHandler.java | 8 ++----
.../falcon/rerun/handler/LateRerunHandler.java | 30 ++++++++++++++++++--
.../falcon/rerun/handler/RetryHandler.java | 17 ++++++++++-
.../falcon/rerun/service/LateRunService.java | 23 ++++++---------
.../falcon/rerun/service/RetryService.java | 9 ++++++
5 files changed, 65 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b9781d12/rerun/src/main/java/org/apache/falcon/rerun/handler/AbstractRerunHandler.java
----------------------------------------------------------------------
diff --git a/rerun/src/main/java/org/apache/falcon/rerun/handler/AbstractRerunHandler.java b/rerun/src/main/java/org/apache/falcon/rerun/handler/AbstractRerunHandler.java
index 124b444..f019737 100644
--- a/rerun/src/main/java/org/apache/falcon/rerun/handler/AbstractRerunHandler.java
+++ b/rerun/src/main/java/org/apache/falcon/rerun/handler/AbstractRerunHandler.java
@@ -24,6 +24,7 @@ import org.apache.falcon.entity.v0.process.Retry;
import org.apache.falcon.rerun.event.RerunEvent;
import org.apache.falcon.rerun.queue.DelayedQueue;
import org.apache.falcon.workflow.WorkflowEngineFactory;
+import org.apache.falcon.workflow.WorkflowExecutionListener;
import org.apache.falcon.workflow.engine.AbstractWorkflowEngine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,7 +35,8 @@ import org.slf4j.LoggerFactory;
* @param <T> a rerun event
* @param <M> queue
*/
-public abstract class AbstractRerunHandler<T extends RerunEvent, M extends DelayedQueue<T>> {
+public abstract class AbstractRerunHandler<T extends RerunEvent, M extends DelayedQueue<T>>
+ implements WorkflowExecutionListener {
protected static final Logger LOG = LoggerFactory.getLogger(LateRerunHandler.class);
protected M delayQueue;
@@ -68,10 +70,6 @@ public abstract class AbstractRerunHandler<T extends RerunEvent, M extends Delay
delayQueue.reconnect();
}
- public Entity getEntity(String entityType, String entityName) throws FalconException {
- return EntityUtil.getEntity(entityType, entityName);
- }
-
public Retry getRetry(Entity entity) throws FalconException {
return EntityUtil.getRetry(entity);
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b9781d12/rerun/src/main/java/org/apache/falcon/rerun/handler/LateRerunHandler.java
----------------------------------------------------------------------
diff --git a/rerun/src/main/java/org/apache/falcon/rerun/handler/LateRerunHandler.java b/rerun/src/main/java/org/apache/falcon/rerun/handler/LateRerunHandler.java
index c1a79fc..76f0ac0 100644
--- a/rerun/src/main/java/org/apache/falcon/rerun/handler/LateRerunHandler.java
+++ b/rerun/src/main/java/org/apache/falcon/rerun/handler/LateRerunHandler.java
@@ -26,7 +26,10 @@ import org.apache.falcon.entity.v0.Entity;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.entity.v0.feed.Feed;
-import org.apache.falcon.entity.v0.process.*;
+import org.apache.falcon.entity.v0.process.Input;
+import org.apache.falcon.entity.v0.process.LateInput;
+import org.apache.falcon.entity.v0.process.LateProcess;
+import org.apache.falcon.entity.v0.process.PolicyType;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.expression.ExpressionHelper;
import org.apache.falcon.hadoop.HadoopClientFactory;
@@ -34,12 +37,13 @@ import org.apache.falcon.rerun.event.LaterunEvent;
import org.apache.falcon.rerun.policy.AbstractRerunPolicy;
import org.apache.falcon.rerun.policy.RerunPolicyFactory;
import org.apache.falcon.rerun.queue.DelayedQueue;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
import org.apache.falcon.workflow.engine.AbstractWorkflowEngine;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import java.util.*;
+import java.util.Date;
/**
* An implementation of handler for late reruns.
@@ -203,4 +207,26 @@ public class LateRerunHandler<M extends DelayedQueue<LaterunEvent>> extends
conf.set(HadoopClientFactory.FS_DEFAULT_NAME_KEY, storageEndpoint);
return conf;
}
+
+ @Override
+ public void onSuccess(WorkflowExecutionContext context) throws FalconException {
+ Entity entity = EntityUtil.getEntity(context.getEntityType(), context.getEntityName());
+ //late data handling not applicable for feed retention action
+ if (context.getOperation() != WorkflowExecutionContext.EntityOperations.DELETE
+ && EntityUtil.getLateProcess(entity) != null) {
+ handleRerun(context.getClusterName(), context.getEntityType(),
+ context.getEntityName(), context.getNominalTimeAsISO8601(),
+ context.getWorkflowRunIdString(), context.getWorkflowId(),
+ context.getWorkflowUser(), context.getExecutionCompletionTime());
+ } else {
+ LOG.info("Late date handling not applicable for entityType: " + context.getEntityType()
+ + ", entityName: " + context.getEntityName()
+ + " operation: " + context.getOperation());
+ }
+ }
+
+ @Override
+ public void onFailure(WorkflowExecutionContext context) throws FalconException {
+ // do nothing since late data does not apply for failed workflows
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b9781d12/rerun/src/main/java/org/apache/falcon/rerun/handler/RetryHandler.java
----------------------------------------------------------------------
diff --git a/rerun/src/main/java/org/apache/falcon/rerun/handler/RetryHandler.java b/rerun/src/main/java/org/apache/falcon/rerun/handler/RetryHandler.java
index 8d8ea03..c6bc36f 100644
--- a/rerun/src/main/java/org/apache/falcon/rerun/handler/RetryHandler.java
+++ b/rerun/src/main/java/org/apache/falcon/rerun/handler/RetryHandler.java
@@ -19,6 +19,7 @@ package org.apache.falcon.rerun.handler;
import org.apache.falcon.FalconException;
import org.apache.falcon.aspect.GenericAlert;
+import org.apache.falcon.entity.EntityUtil;
import org.apache.falcon.entity.v0.Entity;
import org.apache.falcon.entity.v0.Frequency;
import org.apache.falcon.entity.v0.process.PolicyType;
@@ -27,6 +28,7 @@ import org.apache.falcon.rerun.event.RetryEvent;
import org.apache.falcon.rerun.policy.AbstractRerunPolicy;
import org.apache.falcon.rerun.policy.RerunPolicyFactory;
import org.apache.falcon.rerun.queue.DelayedQueue;
+import org.apache.falcon.workflow.WorkflowExecutionContext;
/**
* An implementation of retry handler that kicks off retries until the
@@ -42,7 +44,7 @@ public class RetryHandler<M extends DelayedQueue<RetryEvent>> extends
public void handleRerun(String clusterName, String entityType, String entityName, String nominalTime,
String runId, String wfId, String workflowUser, long msgReceivedTime) {
try {
- Entity entity = getEntity(entityType, entityName);
+ Entity entity = EntityUtil.getEntity(entityType, entityName);
Retry retry = getRetry(entity);
if (retry == null) {
@@ -89,4 +91,17 @@ public class RetryHandler<M extends DelayedQueue<RetryEvent>> extends
daemon.start();
LOG.info("RetryHandler thread started.");
}
+
+ @Override
+ public void onSuccess(WorkflowExecutionContext context) throws FalconException {
+ // do nothing since retry does not apply for failed workflows
+ }
+
+ @Override
+ public void onFailure(WorkflowExecutionContext context) throws FalconException {
+ handleRerun(context.getClusterName(), context.getEntityType(),
+ context.getEntityName(), context.getNominalTimeAsISO8601(),
+ context.getWorkflowRunIdString(), context.getWorkflowId(),
+ context.getWorkflowUser(), context.getExecutionCompletionTime());
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b9781d12/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java
----------------------------------------------------------------------
diff --git a/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java b/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java
index 884a8e2..01563b4 100644
--- a/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java
+++ b/rerun/src/main/java/org/apache/falcon/rerun/service/LateRunService.java
@@ -24,12 +24,12 @@ import org.apache.falcon.rerun.handler.AbstractRerunHandler;
import org.apache.falcon.rerun.handler.RerunHandlerFactory;
import org.apache.falcon.rerun.queue.ActiveMQueue;
import org.apache.falcon.service.FalconService;
+import org.apache.falcon.service.Services;
import org.apache.falcon.util.StartupProperties;
+import org.apache.falcon.workflow.WorkflowJobEndNotificationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.File;
-
/**
* A service implementation for Late Rerun initialized at startup.
*/
@@ -44,6 +44,10 @@ public class LateRunService implements FalconService {
@Override
public void init() throws FalconException {
+ if (!Services.get().isRegistered(WorkflowJobEndNotificationService.NAME)) {
+ throw new FalconException("WorkflowJobEndNotificationService must be configured ahead");
+ }
+
AbstractRerunHandler<LaterunEvent, ActiveMQueue<LaterunEvent>> rerunHandler =
RerunHandlerFactory.getRerunHandler(RerunType.LATE);
ActiveMQueue<LaterunEvent> queue = new ActiveMQueue<LaterunEvent>(
@@ -51,22 +55,13 @@ public class LateRunService implements FalconService {
.getProperty("broker.url", "failover:(tcp://localhost:61616)?initialReconnectDelay=5000"),
"falcon.late.queue");
rerunHandler.init(queue);
+
+ Services.get().<WorkflowJobEndNotificationService>getService(
+ WorkflowJobEndNotificationService.NAME).registerListener(rerunHandler);
}
@Override
public void destroy() throws FalconException {
LOG.info("LateRun thread destroyed");
}
-
- private File getBasePath() {
- File basePath = new File(StartupProperties.get().getProperty(
- "rerun.recorder.path", "/tmp/falcon/rerun"));
- if ((!basePath.exists() && !basePath.mkdirs())
- || (basePath.exists() && !basePath.canWrite())) {
- throw new RuntimeException("Unable to initialize late recorder @"
- + basePath);
- }
-
- return basePath;
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b9781d12/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java
----------------------------------------------------------------------
diff --git a/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java b/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java
index bc64ef5..77bc9d5 100644
--- a/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java
+++ b/rerun/src/main/java/org/apache/falcon/rerun/service/RetryService.java
@@ -25,7 +25,9 @@ import org.apache.falcon.rerun.handler.RerunHandlerFactory;
import org.apache.falcon.rerun.queue.DelayedQueue;
import org.apache.falcon.rerun.queue.InMemoryQueue;
import org.apache.falcon.service.FalconService;
+import org.apache.falcon.service.Services;
import org.apache.falcon.util.StartupProperties;
+import org.apache.falcon.workflow.WorkflowJobEndNotificationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,10 +47,17 @@ public class RetryService implements FalconService {
@Override
public void init() throws FalconException {
+ if (!Services.get().isRegistered(WorkflowJobEndNotificationService.NAME)) {
+ throw new FalconException("WorkflowJobEndNotificationService must be configured ahead");
+ }
+
AbstractRerunHandler<RetryEvent, DelayedQueue<RetryEvent>> rerunHandler =
RerunHandlerFactory.getRerunHandler(RerunType.RETRY);
InMemoryQueue<RetryEvent> queue = new InMemoryQueue<RetryEvent>(getBasePath());
rerunHandler.init(queue);
+
+ Services.get().<WorkflowJobEndNotificationService>getService(
+ WorkflowJobEndNotificationService.NAME).registerListener(rerunHandler);
}
@Override