[OODT-979] Removed XML RPC versions of workflow manager client from all possible
places and added client.close() wherever necessary. Now the WM client CLI
is working without any hanging.
This commit is the last of many which were to stabilize Avro RPC implementations.
diff --git a/pcs/core/src/main/java/org/apache/oodt/pcs/util/WorkflowManagerUtils.java b/pcs/core/src/main/java/org/apache/oodt/pcs/util/WorkflowManagerUtils.java
index e74979d..5db8e05 100644
--- a/pcs/core/src/main/java/org/apache/oodt/pcs/util/WorkflowManagerUtils.java
+++ b/pcs/core/src/main/java/org/apache/oodt/pcs/util/WorkflowManagerUtils.java
@@ -17,10 +17,8 @@
package org.apache.oodt.pcs.util;
-//JDK imports
import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
import org.apache.oodt.cas.workflow.system.rpc.RpcCommunicationFactory;
import org.apache.xmlrpc.XmlRpcClient;
@@ -32,8 +30,6 @@
import java.util.logging.Level;
import java.util.logging.Logger;
-//OODT imports
-
/**
*
* A set of utility methods that can be used by PCS that need to communicate
@@ -48,8 +44,7 @@
private WorkflowManagerClient client;
/* our log stream */
- private static final Logger LOG = Logger.getLogger(WorkflowManagerUtils.class
- .getName());
+ private static final Logger LOG = Logger.getLogger(WorkflowManagerUtils.class.getName());
private URL wmUrl;
diff --git a/webapp/wmservices/src/main/java/org/apache/oodt/cas/wmservices/resources/AbstractWorkflowServiceResource.java b/webapp/wmservices/src/main/java/org/apache/oodt/cas/wmservices/resources/AbstractWorkflowServiceResource.java
index 3d28238..c92e73e 100644
--- a/webapp/wmservices/src/main/java/org/apache/oodt/cas/wmservices/resources/AbstractWorkflowServiceResource.java
+++ b/webapp/wmservices/src/main/java/org/apache/oodt/cas/wmservices/resources/AbstractWorkflowServiceResource.java
@@ -25,11 +25,11 @@
import javax.ws.rs.core.Context;
import org.apache.oodt.cas.wmservices.servlets.WmServicesServlet;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
+import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
public abstract class AbstractWorkflowServiceResource {
- private static final Logger LOGGER = Logger.getLogger(AbstractWorkflowServiceResource.class
- .getName());
+
+ private static final Logger LOGGER = Logger.getLogger(AbstractWorkflowServiceResource.class.getName());
// Servlet context
@Context
@@ -42,10 +42,8 @@
* if an object cannot be retrieved from the context attribute
*/
public File getContextPkgReposDir() throws Exception {
- Object repositoryDirObject = context
- .getAttribute(WmServicesServlet.ATTR_NAME_PKG_REPO_DIR);
- if (repositoryDirObject != null
- && repositoryDirObject instanceof File) {
+ Object repositoryDirObject = context.getAttribute(WmServicesServlet.ATTR_NAME_PKG_REPO_DIR);
+ if (repositoryDirObject instanceof File) {
return (File) repositoryDirObject;
}
String message = "Unable to retrieve packaged repository directory from the servlet context.";
@@ -61,13 +59,11 @@
* @throws Exception
* if an object cannot be retrieved from the context attribute
*/
- public XmlRpcWorkflowManagerClient getContextClient() throws Exception {
+ public WorkflowManagerClient getContextClient() throws Exception {
// Get the workflow manager client from the servlet context.
- Object clientObject = context
- .getAttribute(WmServicesServlet.ATTR_NAME_CLIENT);
- if (clientObject != null
- && clientObject instanceof XmlRpcWorkflowManagerClient) {
- return (XmlRpcWorkflowManagerClient) clientObject;
+ Object clientObject = context.getAttribute(WmServicesServlet.ATTR_NAME_CLIENT);
+ if (clientObject instanceof WorkflowManagerClient) {
+ return (WorkflowManagerClient) clientObject;
}
String message = "Unable to retrieve workflow manager client from the "
diff --git a/webapp/wmservices/src/main/java/org/apache/oodt/cas/wmservices/servlets/WmServicesServlet.java b/webapp/wmservices/src/main/java/org/apache/oodt/cas/wmservices/servlets/WmServicesServlet.java
index 1884d22..4d63daa 100644
--- a/webapp/wmservices/src/main/java/org/apache/oodt/cas/wmservices/servlets/WmServicesServlet.java
+++ b/webapp/wmservices/src/main/java/org/apache/oodt/cas/wmservices/servlets/WmServicesServlet.java
@@ -17,19 +17,20 @@
package org.apache.oodt.cas.wmservices.servlets;
-import java.io.File;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.logging.Level;
-import java.util.logging.Logger;
+import org.apache.cxf.jaxrs.servlet.CXFNonSpringJaxrsServlet;
+import org.apache.oodt.cas.metadata.util.PathUtils;
+import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
+import org.apache.oodt.cas.workflow.system.rpc.RpcCommunicationFactory;
import javax.servlet.ServletConfig;
import javax.servlet.ServletContext;
import javax.servlet.ServletException;
-
-import org.apache.cxf.jaxrs.servlet.CXFNonSpringJaxrsServlet;
-import org.apache.oodt.cas.metadata.util.PathUtils;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
+import java.io.File;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.logging.Level;
+import java.util.logging.Logger;
/**
* Initialize workflow manager services servlet
@@ -43,8 +44,8 @@
* @author vratnakar
*/
public class WmServicesServlet extends CXFNonSpringJaxrsServlet {
- private static final Logger LOGGER = Logger
- .getLogger(WmServicesServlet.class.getName());
+
+ private static final Logger LOGGER = Logger.getLogger(WmServicesServlet.class.getName());
private static final long serialVersionUID = -7830210280506307805L;
@@ -57,7 +58,7 @@
/**
* The name of the servlet context attribute that holds a client for the
- * workflow manager, a {@link XmlRpcWorkflowManagerClient} object.
+ * workflow manager, a {@link org.apache.oodt.cas.workflow.system.WorkflowManagerClient} object.
*/
public static final String ATTR_NAME_CLIENT = "client";
@@ -66,8 +67,9 @@
* packaged repository directory: a {@link File} object.
*/
public static final String ATTR_NAME_PKG_REPO_DIR = "pkgRepoFilesDir";
-
-
+
+ private WorkflowManagerClient client;
+
@Override
public void init(ServletConfig configuration) throws ServletException {
super.init(configuration);
@@ -89,11 +91,10 @@
}
// Attempt to connect the client to the workflow manager and if successful
// store the client as a context attribute for other objects to access.
- XmlRpcWorkflowManagerClient client = new XmlRpcWorkflowManagerClient(url);
+ client = RpcCommunicationFactory.createClient(url);
context.setAttribute(ATTR_NAME_CLIENT, client);
} catch (MalformedURLException e) {
- LOGGER.log(Level.SEVERE,
- "Encountered a malformed URL for the workflow manager.", e);
+ LOGGER.log(Level.SEVERE, "Encountered a malformed URL for the workflow manager.", e);
throw new ServletException(e);
}
@@ -111,4 +112,18 @@
}
}
}
+
+ @Override
+ public void destroy() {
+ if (client != null) {
+ try {
+ client.close();
+ client = null;
+ } catch (IOException e) {
+ LOGGER.severe(String.format("Unable to close WM Client: %s", e.getMessage()));
+ }
+ }
+
+ super.destroy();
+ }
}
diff --git a/workflow/pom.xml b/workflow/pom.xml
index ea0d6b2..47ad334 100644
--- a/workflow/pom.xml
+++ b/workflow/pom.xml
@@ -233,7 +233,7 @@
</descriptors>
<archive>
<manifest>
- <mainClass>org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient</mainClass>
+ <mainClass>org.apache.oodt.cas.workflow.system.AvroRpcWorkflowManagerClient</mainClass>
</manifest>
</archive>
</configuration>
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/DynWorkflowCliAction.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/DynWorkflowCliAction.java
index 0e41165..61fa03d 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/DynWorkflowCliAction.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/DynWorkflowCliAction.java
@@ -16,55 +16,58 @@
*/
package org.apache.oodt.cas.workflow.cli.action;
-//JDK imports
-import java.util.List;
-
-//Apache imports
import org.apache.commons.lang.Validate;
-
-//OODT imports
import org.apache.oodt.cas.cli.exception.CmdLineActionException;
import org.apache.oodt.cas.metadata.Metadata;
+import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
+
+import java.util.List;
+import java.util.logging.Logger;
+
/**
* A {@link CmdLineAction} for submitting dynamically created {@link Workflow}s
* of pre-defined {@link WorkflowTask}s.
- *
+ *
* @author bfoster (Brian Foster)
*/
public class DynWorkflowCliAction extends WorkflowCliAction {
- private List<String> taskIds;
- private Metadata metadata;
+ private static final Logger LOGGER = Logger.getLogger(DynWorkflowCliAction.class.getName());
- public DynWorkflowCliAction() {
- metadata = new Metadata();
- }
+ private List<String> taskIds;
+ private Metadata metadata;
- @Override
- public void execute(ActionMessagePrinter printer)
- throws CmdLineActionException {
- Validate.notNull(taskIds, "Must specify taskIds");
+ public DynWorkflowCliAction() {
+ metadata = new Metadata();
+ }
- try {
- String instId = getClient().executeDynamicWorkflow(taskIds, metadata);
- printer.println("Started dynamic workflow with id '" + instId + "'");
- } catch (Exception e) {
- throw new CmdLineActionException(
- "Failed to submit dynamic workflow for taskIds " + taskIds
- + " with metadata " + metadata.getMap() + " : "
- + e.getMessage(), e);
- }
- }
+ @Override
+ public void execute(ActionMessagePrinter printer)
+ throws CmdLineActionException {
+ Validate.notNull(taskIds, "Must specify taskIds");
- public void setTaskIds(List<String> taskIds) {
- this.taskIds = taskIds;
- }
+ try (WorkflowManagerClient client = getClient()) {
+ LOGGER.fine(String.format("Starting workflow %d tasks", taskIds.size()));
+ String instId = client.executeDynamicWorkflow(taskIds, metadata);
+ LOGGER.fine(String.format("Started workflow with instanceId: %s", instId));
+ printer.println("Started dynamic workflow with id '" + instId + "'");
+ } catch (Exception e) {
+ throw new CmdLineActionException(
+ "Failed to submit dynamic workflow for taskIds " + taskIds
+ + " with metadata " + metadata.getMap() + " : "
+ + e.getMessage(), e);
+ }
+ }
- public void addMetadata(List<String> metadata) {
- Validate.isTrue(metadata.size() > 1);
+ public void setTaskIds(List<String> taskIds) {
+ this.taskIds = taskIds;
+ }
- this.metadata.addMetadata(metadata.get(0),
- metadata.subList(1, metadata.size()));
- }
+ public void addMetadata(List<String> metadata) {
+ Validate.isTrue(metadata.size() > 1);
+
+ this.metadata.addMetadata(metadata.get(0),
+ metadata.subList(1, metadata.size()));
+ }
}
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetConditionByIdCliAction.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetConditionByIdCliAction.java
index e5d16b1..f9c0b41 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetConditionByIdCliAction.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetConditionByIdCliAction.java
@@ -19,6 +19,7 @@
//OODT imports
import org.apache.oodt.cas.cli.exception.CmdLineActionException;
import org.apache.oodt.cas.workflow.structs.WorkflowCondition;
+import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
/**
* A {@link CmdLineAction} which retrieves WorkflowCondition information for
@@ -33,9 +34,10 @@
@Override
public void execute(ActionMessagePrinter printer)
throws CmdLineActionException {
+
try {
- WorkflowCondition condition = getClient()
- .getConditionById(conditionId);
+ WorkflowManagerClient client = getClient();
+ WorkflowCondition condition = client.getConditionById(conditionId);
printer.println("Condition: [id=" + condition.getConditionId()
+ ", name=" + condition.getConditionName() + ", order="
+ condition.getOrder() + ", class="
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetFirstPageCliAction.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetFirstPageCliAction.java
index 60f9c99..f8431b6 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetFirstPageCliAction.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetFirstPageCliAction.java
@@ -24,7 +24,6 @@
import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
import org.apache.oodt.cas.workflow.structs.WorkflowInstancePage;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
/**
* Gets the first page of workflow instances.
@@ -66,7 +65,6 @@
+ client.getWorkflowCurrentTaskWallClockMinutes(inst.getId())
+ "]");
}
-
} catch (Exception e) {
throw new CmdLineActionException(
"Failed to get first page of workflows : " + e.getMessage(), e);
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetLastPageCliAction.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetLastPageCliAction.java
index f1ccd18..75e51ce 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetLastPageCliAction.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetLastPageCliAction.java
@@ -24,10 +24,9 @@
import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
import org.apache.oodt.cas.workflow.structs.WorkflowInstancePage;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
/**
- * A {@link CmdLineOption} which gets the last page of workflows.
+ * A {@link org.apache.oodt.cas.cli.option.CmdLineOption} which gets the last page of workflows.
*
* @author bfoster (Brian Foster)
*/
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetNextPageCliAction.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetNextPageCliAction.java
index e2babe4..6e6a19b 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetNextPageCliAction.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetNextPageCliAction.java
@@ -27,11 +27,10 @@
import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
import org.apache.oodt.cas.workflow.structs.WorkflowInstancePage;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
/**
* A {@link CmdLineAction} which gets the next page of workflows.
- *
+ *
* @author bfoster (Brian Foster)
*/
public class GetNextPageCliAction extends WorkflowCliAction {
@@ -41,8 +40,7 @@
@SuppressWarnings("unchecked")
@Override
- public void execute(ActionMessagePrinter printer)
- throws CmdLineActionException {
+ public void execute(ActionMessagePrinter printer) throws CmdLineActionException {
Validate.isTrue(pageNum != -1);
try {
@@ -61,8 +59,7 @@
throw new Exception(
"WorkflowManager returned null page of workflows");
}
- for (WorkflowInstance inst : (List<WorkflowInstance>) page
- .getPageWorkflows()) {
+ for (WorkflowInstance inst : (List<WorkflowInstance>) page.getPageWorkflows()) {
printer.println("Instance: [id=" + inst.getId() + ", status="
+ inst.getStatus() + ", currentTask="
+ inst.getCurrentTaskId() + ", workflow="
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetPrevPageCliAction.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetPrevPageCliAction.java
index c4cd9bc..ab006d3 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetPrevPageCliAction.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/GetPrevPageCliAction.java
@@ -27,10 +27,9 @@
import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
import org.apache.oodt.cas.workflow.structs.WorkflowInstancePage;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
/**
- * A {@link CmdLineAction} which gets the previous page of workflows.
+ * A {@link org.apache.oodt.cas.cli.action.CmdLineAction} which gets the previous page of workflows.
*
* @author bfoster (Brian Foster)
*/
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/WorkflowCliAction.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/WorkflowCliAction.java
index 84295a0..5673aec 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/WorkflowCliAction.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/cli/action/WorkflowCliAction.java
@@ -16,18 +16,15 @@
*/
package org.apache.oodt.cas.workflow.cli.action;
-//JDK imports
-import java.net.MalformedURLException;
-import java.net.URL;
-//Apache imports
import org.apache.commons.lang.Validate;
-
-//OODT imports
import org.apache.oodt.cas.cli.action.CmdLineAction;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
import org.apache.oodt.cas.workflow.system.rpc.RpcCommunicationFactory;
+import java.net.MalformedURLException;
+import java.net.URL;
+
/**
* Base {@link CmdLineAction} for Workflow Manager.
*
@@ -35,24 +32,37 @@
*/
public abstract class WorkflowCliAction extends CmdLineAction {
- private WorkflowManagerClient client;
+ private WorkflowManagerClient client;
- public String getUrl() {
- return System.getProperty("org.apache.oodt.cas.workflow.url");
- }
+ public String getUrl() {
+ return System.getProperty("org.apache.oodt.cas.workflow.url");
+ }
- protected WorkflowManagerClient getClient()
- throws MalformedURLException {
- Validate.notNull(getUrl());
+ protected synchronized WorkflowManagerClient getClient() throws MalformedURLException {
+ Validate.notNull(getUrl());
- if (client != null) {
- return client;
- } else {
- return RpcCommunicationFactory.createClient(new URL(getUrl()));
- }
- }
+ if (client == null) {
+ client = RpcCommunicationFactory.createClient(new URL(getUrl()));
+ }
- public void setClient(WorkflowManagerClient client) {
- this.client = client;
- }
+ return client;
+ }
+
+ public void setClient(WorkflowManagerClient client) {
+ this.client = client;
+ }
+
+ /**
+ * This is not the best way to close the client. For the time being, we go with this way.
+ *
+ * @throws Throwable
+ */
+ @Override
+ public void finalize() throws Throwable {
+ if (client != null) {
+ client.close();
+ }
+
+ super.finalize();
+ }
}
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/examples/BranchRedirector.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/examples/BranchRedirector.java
index 9ae14d8..0bc286e 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/examples/BranchRedirector.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/examples/BranchRedirector.java
@@ -27,12 +27,11 @@
import org.apache.oodt.cas.workflow.structs.WorkflowTaskInstance;
import org.apache.oodt.cas.workflow.structs.exceptions.WorkflowTaskInstanceException;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
import org.apache.oodt.cas.workflow.system.rpc.RpcCommunicationFactory;
/**
*
- * Redirects from an existing {@link WorkflowInstance} by sending a specified
+ * Redirects from an existing {@link org.apache.oodt.cas.workflow.structs.WorkflowInstance} by sending a specified
* event specified by the task configuration parameter named
* <code>eventName</code>.
*
@@ -56,15 +55,12 @@
@Override
public void run(Metadata metadata, WorkflowTaskConfiguration config)
throws WorkflowTaskInstanceException {
- WorkflowManagerClient wm;
- try {
- wm = RpcCommunicationFactory.createClient(new URL(
- metadata.getMetadata(CoreMetKeys.WORKFLOW_MANAGER_URL)));
+ try (WorkflowManagerClient wm = RpcCommunicationFactory
+ .createClient(new URL(metadata.getMetadata(CoreMetKeys.WORKFLOW_MANAGER_URL)))) {
wm.sendEvent(config.getProperty("eventName"), metadata);
} catch (Exception e) {
throw new WorkflowTaskInstanceException(e.getMessage());
}
}
-
}
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/examples/RandomStatusUpdateTask.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/examples/RandomStatusUpdateTask.java
index a1dd885..af0a933 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/examples/RandomStatusUpdateTask.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/examples/RandomStatusUpdateTask.java
@@ -19,13 +19,11 @@
package org.apache.oodt.cas.workflow.examples;
-//OODT imports
-
import org.apache.oodt.cas.metadata.Metadata;
+import org.apache.oodt.cas.workflow.struct.avrotypes.AvroWorkflowInstance;
import org.apache.oodt.cas.workflow.structs.WorkflowTaskConfiguration;
import org.apache.oodt.cas.workflow.structs.WorkflowTaskInstance;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
import org.apache.oodt.cas.workflow.system.rpc.RpcCommunicationFactory;
import java.net.MalformedURLException;
@@ -34,15 +32,13 @@
import java.util.logging.Level;
import java.util.logging.Logger;
-//JDK imports
-
/**
* @author mattmann
* @version $Revision$
*
* This class illustrates OODT-86, demonstrating how the method
- * {@link org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManager#updateWorkflowInstance(java.util.Hashtable)}
+ * {@link org.apache.oodt.cas.workflow.system.AvroRpcWorkflowManager#updateWorkflowInstance(AvroWorkflowInstance)}
* allows a user to change the status of a given {@link org.apache.oodt.cas.workflow.structs.WorkflowInstance}
* programmatically.
*/
@@ -110,4 +106,10 @@
}
}
+ @Override
+ public void finalize() throws Exception {
+ if (client != null) {
+ client.close();
+ }
+ }
}
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/structs/RollbackableWorkflowTaskInstance.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/structs/RollbackableWorkflowTaskInstance.java
index 3759570..d2b3929 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/structs/RollbackableWorkflowTaskInstance.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/structs/RollbackableWorkflowTaskInstance.java
@@ -1,75 +1,72 @@
package org.apache.oodt.cas.workflow.structs;
//JDK imports
+import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.List;
+import java.util.logging.Logger;
//OODT imports
import org.apache.oodt.cas.metadata.Metadata;
import org.apache.oodt.cas.workflow.metadata.CoreMetKeys;
import org.apache.oodt.cas.workflow.structs.exceptions.WorkflowTaskInstanceException;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
+import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
+import org.apache.oodt.cas.workflow.system.rpc.RpcCommunicationFactory;
-public abstract class RollbackableWorkflowTaskInstance implements
- WorkflowTaskInstance {
-
- protected XmlRpcWorkflowManagerClient wmc;
+public abstract class RollbackableWorkflowTaskInstance implements WorkflowTaskInstance {
+
+ private static final Logger LOGGER= Logger.getLogger(RollbackableWorkflowTaskInstance.class.getName());
+
protected String workflowInstId;
-
- public void run(Metadata metadata, WorkflowTaskConfiguration config)
- throws WorkflowTaskInstanceException {
-
+
+ public void run(Metadata metadata, WorkflowTaskConfiguration config) throws WorkflowTaskInstanceException {
workflowInstId = getWorkflowInstanceId(metadata);
- wmc = createWorkflowMangerClient(metadata);
-
-
- //clean generated metadata
- clearAllMetadata(metadata);
-
- //Determine next workflow state
- updateState(metadata);
-
- //clean up task instance from instance rep?
- clearInstRep();
+ try (WorkflowManagerClient wmc = createWorkflowMangerClient(metadata)){
+ //clean generated metadata
+ clearAllMetadata(wmc, metadata);
+
+ //Determine next workflow state
+ updateState(metadata);
+
+ //clean up task instance from instance rep?
+ clearInstRep();
+ } catch (IOException e) {
+ LOGGER.severe(String.format("WorkflowManagerClient error : %s", e.getMessage()));
+ }
+ }
+
+ private String getWorkflowInstanceId(Metadata metadata) {
+ return metadata.getMetadata(CoreMetKeys.WORKFLOW_INST_ID);
}
-
- protected String getWorkflowInstanceId(Metadata metadata) {
-
- String instId = metadata.getMetadata(CoreMetKeys.WORKFLOW_INST_ID);
- return instId;
-
- }
-
- protected XmlRpcWorkflowManagerClient createWorkflowMangerClient(Metadata metadata) {
+
+ private WorkflowManagerClient createWorkflowMangerClient(Metadata metadata) {
URL url;
try {
url = new URL(metadata.getMetadata(CoreMetKeys.WORKFLOW_MANAGER_URL));
- wmc = new XmlRpcWorkflowManagerClient(url);
-
+ return RpcCommunicationFactory.createClient(url);
} catch (MalformedURLException e) {
e.printStackTrace();
}
- return wmc;
+ return null;
}
- protected void clearAllMetadata(Metadata metadata) {
-
+ private void clearAllMetadata(WorkflowManagerClient client, Metadata metadata) {
List<String> keys = metadata.getAllKeys();
-
+
//remove metadata for each key
for(String key: keys) {
metadata.removeMetadata(key);
- }
+ }
try {
- wmc.updateMetadataForWorkflow(workflowInstId, metadata);
+ client.updateMetadataForWorkflow(workflowInstId, metadata);
} catch (Exception e) {
e.printStackTrace();
}
}
-
+
public abstract void updateState(Metadata metadata);
-
+
public abstract void clearInstRep();
}
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/structs/TaskJob.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/structs/TaskJob.java
index c10c8ef..168bb53 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/structs/TaskJob.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/structs/TaskJob.java
@@ -118,9 +118,7 @@
private void updateStatus(String status, Metadata met) {
String workflowInstId = met.getMetadata(WORKFLOW_INST_ID);
- WorkflowManagerClient wClient = getWmClientFromMetadata(met);
-
- try {
+ try (WorkflowManagerClient wClient = getWmClientFromMetadata(met)) {
if (!wClient.updateWorkflowInstanceStatus(workflowInstId, status)) {
LOG.log(Level.WARNING,
"Unable to update status for workflow instance: ["
@@ -132,9 +130,8 @@
private void updateMetadata(Metadata met) {
String workflowInstId = met.getMetadata(WORKFLOW_INST_ID);
- WorkflowManagerClient wClient = getWmClientFromMetadata(met);
- try {
+ try (WorkflowManagerClient wClient = getWmClientFromMetadata(met)) {
if (!wClient.updateMetadataForWorkflow(workflowInstId, met)) {
LOG.log(Level.WARNING,
"Unable to update Metadata for workflow instance: ["
@@ -147,9 +144,8 @@
private void setWorkflowInstanceCurrentTaskStartDateTime(
String startDateTime, Metadata met) {
String workflowInstId = met.getMetadata(WORKFLOW_INST_ID);
- WorkflowManagerClient wClient = getWmClientFromMetadata(met);
- try {
+ try (WorkflowManagerClient wClient = getWmClientFromMetadata(met)) {
if (!wClient.setWorkflowInstanceCurrentTaskStartDateTime(
workflowInstId, startDateTime)) {
LOG.log(Level.WARNING,
@@ -163,9 +159,8 @@
private void setWorkflowInstanceCurrentTaskEndDateTime(String endDateTime,
Metadata met) {
String workflowInstId = met.getMetadata(WORKFLOW_INST_ID);
- WorkflowManagerClient wClient = getWmClientFromMetadata(met);
- try {
+ try (WorkflowManagerClient wClient = getWmClientFromMetadata(met)) {
if (!wClient.setWorkflowInstanceCurrentTaskEndDateTime(
workflowInstId, endDateTime)) {
LOG.log(Level.WARNING,
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/system/WorkflowManager.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/system/WorkflowManager.java
index 7453e78..eda3027 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/system/WorkflowManager.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/system/WorkflowManager.java
@@ -1,26 +1,17 @@
package org.apache.oodt.cas.workflow.system;
-import org.apache.oodt.cas.workflow.structs.exceptions.EngineException;
-import org.apache.oodt.cas.workflow.structs.exceptions.InstanceRepositoryException;
-import org.apache.oodt.cas.workflow.structs.exceptions.RepositoryException;
-
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.util.Hashtable;
-import java.util.Vector;
-
/**
* @author radu
*
* Workflow manager interface, used for RPC implementatino.
*/
-public interface WorkflowManager{
+public interface WorkflowManager {
- public static final String PROPERTIES_FILE_PROPERTY = "org.apache.oodt.cas.workflow.properties";
- public static final String WORKFLOW_ENGINE_FACTORY_PROPERTY = "workflow.engine.factory";
- public static final String ENGINE_RUNNER_FACTORY_PROPERTY = "workflow.engine.runner.factory";
- public static final String WORKFLOW_REPOSITORY_FACTORY_PROPERTY = "workflow.repo.factory";
- public static final int DEFAULT_WEB_SERVER_PORT = 9001;
+ String PROPERTIES_FILE_PROPERTY = "org.apache.oodt.cas.workflow.properties";
+ String WORKFLOW_ENGINE_FACTORY_PROPERTY = "workflow.engine.factory";
+ String ENGINE_RUNNER_FACTORY_PROPERTY = "workflow.engine.runner.factory";
+ String WORKFLOW_REPOSITORY_FACTORY_PROPERTY = "workflow.repo.factory";
+ int DEFAULT_WEB_SERVER_PORT = 9001;
/**
*
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/system/XmlRpcWorkflowManagerClient.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/system/XmlRpcWorkflowManagerClient.java
index f9bc6e6..4d05b3f 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/system/XmlRpcWorkflowManagerClient.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/system/XmlRpcWorkflowManagerClient.java
@@ -18,7 +18,6 @@
package org.apache.oodt.cas.workflow.system;
-//APACHE imports
import org.apache.oodt.cas.metadata.Metadata;
import org.apache.oodt.cas.workflow.structs.Workflow;
@@ -33,15 +32,12 @@
import java.io.IOException;
import java.net.URL;
-import java.util.Map;
import java.util.List;
+import java.util.Map;
import java.util.Vector;
import java.util.logging.Level;
import java.util.logging.Logger;
-//JDK imports
-//OODT imports
-
/**
* @author mattmann
* @version $Revision$
@@ -49,7 +45,7 @@
* <p>
* The XML RPC based workflow manager client.
* </p>
- *
+ *
*/
@Deprecated
public class XmlRpcWorkflowManagerClient implements WorkflowManagerClient {
@@ -69,7 +65,7 @@
* Constructs a new XmlRpcWorkflowManagerClient with the given
* <code>url</code>.
* </p>
- *
+ *
* @param url
* The url pointer to the xml rpc workflow manager service.
*/
@@ -77,7 +73,7 @@
client = new XmlRpcClient(url);
workflowManagerUrl = url;
}
-
+
@Override
public boolean refreshRepository()
throws XmlRpcException, IOException {
@@ -93,7 +89,7 @@
Vector<String> taskIdVector = new Vector<String>();
taskIdVector.addAll(taskIds);
String instId;
-
+
argList.add(taskIdVector);
argList.add(metadata.getHashTable());
@@ -368,10 +364,10 @@
.execute("workflowmgr.handleEvent", argList);
}
-
+
public boolean clearWorkflowInstances() throws XmlRpcException, IOException{
Vector argList = new Vector();
-
+
return (Boolean)client.execute("workflowmgr.clearWorkflowInstances", argList);
}
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/system/rpc/RpcCommunicationFactory.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/system/rpc/RpcCommunicationFactory.java
index f0b89ac..244fe8f 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/system/rpc/RpcCommunicationFactory.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/system/rpc/RpcCommunicationFactory.java
@@ -46,7 +46,7 @@
e.printStackTrace();
}
return properties.getProperty("workflow.server.factory",
- "org.apache.oodt.cas.workflow.system.rpc.XmlRpcWorkflowManagerFactory");
+ "org.apache.oodt.cas.workflow.system.rpc.AvroRpcWorkflowManagerFactory");
}
private static String getRpcClientClassName() {
@@ -58,7 +58,7 @@
logger.error("Unable to load properties", e);
}
return properties.getProperty("workflow.client.factory",
- "org.apache.oodt.cas.workflow.system.rpc.XmlRpcWorkflowManagerFactory");
+ "org.apache.oodt.cas.workflow.system.rpc.AvroRpcWorkflowManagerFactory");
}
public static WorkflowManager createServer(int port) {
@@ -66,6 +66,7 @@
WorkflowManagerFactory workflowManagerFactory =
(WorkflowManagerFactory) Class.forName(getRpcServerClassName()).newInstance();
workflowManagerFactory.setPort(port);
+ logger.debug("Using workflow manager server factory : {}", workflowManagerFactory.getClass());
return workflowManagerFactory.createServer();
} catch (Exception e) {
logger.error("Error creating server", e);
@@ -78,6 +79,7 @@
WorkflowManagerFactory workflowManagerFactory =
(WorkflowManagerFactory) Class.forName(getRpcClientClassName()).newInstance();
workflowManagerFactory.setUrl(url);
+ logger.debug("Using workflow manager client factory : {}", workflowManagerFactory.getClass());
return workflowManagerFactory.createClient();
} catch (Exception e) {
logger.error("Unable to create client", e);
diff --git a/workflow/src/main/java/org/apache/oodt/cas/workflow/tools/InstanceRepoCleaner.java b/workflow/src/main/java/org/apache/oodt/cas/workflow/tools/InstanceRepoCleaner.java
index de2d6f5..46e8b48 100644
--- a/workflow/src/main/java/org/apache/oodt/cas/workflow/tools/InstanceRepoCleaner.java
+++ b/workflow/src/main/java/org/apache/oodt/cas/workflow/tools/InstanceRepoCleaner.java
@@ -17,17 +17,15 @@
package org.apache.oodt.cas.workflow.tools;
-//JDK imports
import org.apache.oodt.cas.workflow.instrepo.LuceneWorkflowInstanceRepository;
import org.apache.oodt.cas.workflow.structs.WorkflowInstance;
import org.apache.oodt.cas.workflow.structs.WorkflowInstancePage;
import org.apache.oodt.cas.workflow.structs.WorkflowStatus;
-import org.apache.oodt.cas.workflow.structs.exceptions.InstanceRepositoryException;
import org.apache.oodt.cas.workflow.system.WorkflowManagerClient;
-import org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManagerClient;
+import org.apache.oodt.cas.workflow.system.rpc.RpcCommunicationFactory;
import org.apache.oodt.commons.date.DateUtils;
-import org.apache.xmlrpc.XmlRpcException;
+import java.io.Closeable;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
@@ -46,7 +44,7 @@
* @author mattmann
*
*/
-public class InstanceRepoCleaner {
+public class InstanceRepoCleaner implements Closeable {
/* PGE task statuses */
public static final String STAGING_INPUT = "STAGING INPUT";
@@ -57,8 +55,7 @@
public static final String CRAWLING = "CRAWLING";
- private static final Logger LOG = Logger.getLogger(InstanceRepoCleaner.class
- .getName());
+ private static final Logger LOG = Logger.getLogger(InstanceRepoCleaner.class.getName());
private WorkflowManagerClient wm;
@@ -68,7 +65,7 @@
}
public InstanceRepoCleaner(String wmUrlStr) throws MalformedURLException {
- this.wm = new XmlRpcWorkflowManagerClient(new URL(wmUrlStr));
+ this.wm = RpcCommunicationFactory.createClient(new URL(wmUrlStr));
}
public void setInstanceRepo(String idxPath) {
@@ -83,30 +80,34 @@
System.exit(1);
}
- InstanceRepoCleaner clean;
- if (args.length == 1) {
- String wmUrlStr = args[0];
- clean = new InstanceRepoCleaner(wmUrlStr);
- } else {
- String idxPath = args[1];
- clean = new InstanceRepoCleaner();
- clean.setInstanceRepo(idxPath);
+ InstanceRepoCleaner clean = null;
+ try {
+ if (args.length == 1) {
+ String wmUrlStr = args[0];
+ clean = new InstanceRepoCleaner(wmUrlStr);
+ } else {
+ String idxPath = args[1];
+ clean = new InstanceRepoCleaner();
+ clean.setInstanceRepo(idxPath);
+ }
+ clean.cleanRepository();
+ } finally {
+ if (clean != null) {
+ clean.close();
+ }
}
- clean.cleanRepository();
}
public void cleanRepository() throws Exception {
- WorkflowInstancePage page = wm != null ? wm.getFirstPage() : rep
- .getFirstPage();
- while (page != null && page.getPageWorkflows() != null
- && page.getPageWorkflows().size() > 0) {
+ WorkflowInstancePage page = wm != null ? wm.getFirstPage() : rep.getFirstPage();
+ while (page != null && page.getPageWorkflows() != null && page.getPageWorkflows().size() > 0) {
LOG.log(Level.INFO,
"Cleaning workflow instances: page: [" + page.getPageNum() + "] of ["
+ page.getTotalPages() + "]: page size: [" + page.getPageSize()
+ "]");
- for (WorkflowInstance inst : (List<WorkflowInstance>) page
- .getPageWorkflows()) {
+
+ for (WorkflowInstance inst : (List<WorkflowInstance>) page.getPageWorkflows()) {
if (inst.getStatus().equals(WorkflowStatus.CREATED)
|| inst.getStatus().equals(WorkflowStatus.STARTED)
|| inst.getStatus().equals(WorkflowStatus.QUEUED)
@@ -140,9 +141,13 @@
}
page = wm != null ? wm.getNextPage(page) : rep.getNextPage(page);
-
}
-
}
+ @Override
+ public void close() throws IOException {
+ if (wm != null) {
+ wm.close();
+ }
+ }
}
