Github user sandeepSamudrala commented on a diff in the pull request:
https://github.com/apache/falcon/pull/17#discussion_r51236461
--- Diff:
common/src/main/java/org/apache/falcon/workflow/WorkflowJobEndNotificationService.java
---
@@ -152,48 +157,48 @@ public void notifyWait(WorkflowExecutionContext
context) {
// The method retrieves the conf from the cache if it is in cache.
// Else, queries WF Engine to retrieve the conf of the workflow
- private void updateContextFromWFConf(WorkflowExecutionContext context)
{
- try {
- Properties wfProps = contextMap.get(context.getWorkflowId());
- if (wfProps == null) {
- Entity entity =
CONFIG_STORE.get(EntityType.valueOf(context.getEntityType()),
context.getEntityName());
- // Entity can be null in case of delete. Engine will
generate notifications for instance kills.
- // But, the entity would no longer be in the config store.
- if (entity == null) {
- return;
- }
- for (String cluster :
EntityUtil.getClustersDefinedInColos(entity)) {
- try {
- InstancesResult.Instance[] instances =
WorkflowEngineFactory.getWorkflowEngine(entity)
- .getJobDetails(cluster,
context.getWorkflowId()).getInstances();
- if (instances != null && instances.length > 0) {
- wfProps =
getWFProps(instances[0].getWfParams());
- // Required by RetryService. But, is not part
of conf.
-
wfProps.setProperty(WorkflowExecutionArgs.RUN_ID.getName(),
-
Integer.toString(instances[0].getRunId()));
- }
- } catch (FalconException e) {
- // Do Nothing. The workflow may not have been
deployed on this cluster.
- continue;
+ private boolean updateContextFromWFConf(WorkflowExecutionContext
context) throws FalconException {
+ Properties wfProps = contextMap.get(context.getWorkflowId());
+ if (wfProps == null) {
+ Entity entity = null;
+ try {
+ entity = EntityUtil.getEntity(context.getEntityType(),
context.getEntityName());
+ } catch (EntityNotRegisteredException e) {
+ // Entity no longer exists. No need to notify.
+ LOG.debug("Entity {} of type {} doesn't exist in config
store. Notification Ignored.",
+ context.getEntityName(), context.getEntityType());
+ contextMap.remove(context.getWorkflowId());
+ return false;
+ }
+ for (String cluster :
EntityUtil.getClustersDefinedInColos(entity)) {
+ try {
+ InstancesResult.Instance[] instances =
WorkflowEngineFactory.getWorkflowEngine(entity)
+ .getJobDetails(cluster,
context.getWorkflowId()).getInstances();
+ if (instances != null && instances.length > 0) {
+ wfProps = getWFProps(instances[0].getWfParams());
+ // Required by RetryService. But, is not part of
conf.
+
wfProps.setProperty(WorkflowExecutionArgs.RUN_ID.getName(),
+ Integer.toString(instances[0].getRunId()));
}
- contextMap.put(context.getWorkflowId(), wfProps);
+ } catch (FalconException e) {
+ // Do Nothing. The workflow may not have been deployed
on this cluster.
--- End diff --
can this state not be caught?for if it was scheduled on this cluster
easily? rather than catching it as an exception?.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
