Github user pallavi-rao commented on a diff in the pull request:
https://github.com/apache/falcon/pull/17#discussion_r51245197
--- 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 --
The comment is wrong. Since we get the clusters for a given entity on that
colo, this should not happen at all. In fact, the whole try-catch block is
superfluous. Will remove.
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
