AMBARI-4401. Schedule_request with failing batch forever stays in state SCHEDULED. (swagle)

ambari-server/src/main/java/org/apache/ambari/server/scheduler/AbstractLinearExecutionJob.java

@@ -73,6 +73,7 @@ public abstract class AbstractLinearExecutionJob implements ExecutionJob {
   public void execute(JobExecutionContext context) throws JobExecutionException {
     JobKey jobKey = context.getJobDetail().getKey();
     LOG.debug("Executing linear job: " + jobKey);
+    JobDataMap jobDataMap = context.getMergedJobDataMap();
 
     if (!executionScheduleManager.continueOnMisfire(context)) {
       throw new JobExecutionException("Canceled execution based on misfire"
@@ -80,29 +81,43 @@ public abstract class AbstractLinearExecutionJob implements ExecutionJob {
         + ", scheduleTime = " + context.getScheduledFireTime());
     }
 
-    Map<String, Object> properties = context.getMergedJobDataMap()
-      .getWrappedMap();
+    Map<String, Object> properties = jobDataMap.getWrappedMap();
+    boolean finalize = false;
 
     // Perform work and exit if failure reported
     try {
       doWork(properties);
     } catch (AmbariException e) {
-      LOG.error("Exception caught on execution of job " + jobKey + ". Exiting linear chain...", e);
+      LOG.error("Exception caught on execution of job " + jobKey +
+        ". Exiting linear chain...", e);
+
+      finalize = true;
       throw new JobExecutionException(e);
+
     } catch (RuntimeException e) {
       LOG.error("Unexpected exception caught on execution of job " + jobKey + ". " +
           "Exiting linear chain...", e);
+
+      finalize = true;
       throw e;
+
+    } finally {
+      // Finalize before exiting chain
+      if (finalize) {
+        try {
+          finalizeExecution(properties);
+        } catch (AmbariException e) {
+          LOG.warn("Unable to finalize execution for job: " + jobKey);
+        }
+      }
     }
 
     LOG.debug("Finished linear job: " + jobKey);
 
-    JobDataMap jobDataMap = context.getMergedJobDataMap();
-
     String nextJobName = jobDataMap.getString(NEXT_EXECUTION_JOB_NAME_KEY);
     String nextJobGroup = jobDataMap.getString(NEXT_EXECUTION_JOB_GROUP_KEY);
 
-    // If no more jobs left, update status and return
+    // If no more jobs left, finalize and return
     if (nextJobName == null || nextJobName.isEmpty()) {
       LOG.debug("End of linear job chain. Returning with success.");
       try {
@@ -125,5 +140,4 @@ public abstract class AbstractLinearExecutionJob implements ExecutionJob {
 
     executionScheduleManager.scheduleJob(trigger);
   }
-
 }

ambari-server/src/test/java/org/apache/ambari/server/scheduler/ExecutionScheduleManagerTest.java

@@ -27,6 +27,7 @@ import com.google.inject.persist.PersistService;
 import com.google.inject.persist.Transactional;
 import com.google.inject.util.Modules;
 import junit.framework.Assert;
+import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.actionmanager.ActionDBAccessor;
 import org.apache.ambari.server.actionmanager.HostRoleCommand;
 import org.apache.ambari.server.actionmanager.HostRoleStatus;
@@ -46,8 +47,10 @@ import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.quartz.CronTrigger;
+import org.quartz.JobDataMap;
 import org.quartz.JobDetail;
 import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
 import org.quartz.JobKey;
 import org.quartz.Scheduler;
 import org.quartz.SchedulerException;
@@ -578,4 +581,41 @@ public class ExecutionScheduleManagerTest {
       executionSchedulerMock, scheduleManager, batchMock, batchRequestMock,
       triggerMock, jobDetailMock, actionDBAccessorMock);
   }
+
+  @Test
+  public void testFinalizeBeforeExit() throws Exception {
+    ExecutionScheduleManager scheduleManagerMock = createMock(ExecutionScheduleManager.class);
+    AbstractLinearExecutionJob executionJob =
+      createMockBuilder(AbstractLinearExecutionJob.class)
+      .addMockedMethods("finalizeExecution", "doWork")
+      .withConstructor(scheduleManagerMock)
+      .createMock();
+    JobExecutionContext context = createMock(JobExecutionContext.class);
+    JobDetail jobDetail = createMock(JobDetail.class);
+    JobDataMap jobDataMap = createMock(JobDataMap.class);
+
+    expect(context.getJobDetail()).andReturn(jobDetail).anyTimes();
+    expect(context.getMergedJobDataMap()).andReturn(jobDataMap).anyTimes();
+    expect(jobDetail.getKey()).andReturn(new JobKey("TestJob"));
+    expect(jobDataMap.getWrappedMap()).andReturn(new HashMap<String,Object>());
+    expect(scheduleManagerMock.continueOnMisfire(context)).andReturn(true);
+
+    executionJob.doWork((Map<String, Object>) anyObject());
+    expectLastCall().andThrow(new AmbariException("Test Exception")).anyTimes();
+
+    executionJob.finalizeExecution((Map<String, Object>) anyObject());
+    expectLastCall().once();
+
+    replay(scheduleManagerMock, executionJob, context, jobDataMap, jobDetail);
+
+    try {
+      executionJob.execute(context);
+    } catch (Exception ae) {
+      assertThat(ae, instanceOf(JobExecutionException.class));
+      JobExecutionException je = (JobExecutionException) ae;
+      Assert.assertEquals("Test Exception", je.getUnderlyingException().getMessage());
+    }
+
+    verify(scheduleManagerMock, executionJob, context, jobDataMap, jobDetail);
+  }
 }

ambari-server/src/test/java/org/apache/ambari/server/state/scheduler/BatchRequestJobTest.java

@@ -18,12 +18,16 @@
 
 package org.apache.ambari.server.state.scheduler;
 
+import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.actionmanager.HostRoleStatus;
+import org.apache.ambari.server.scheduler.AbstractLinearExecutionJob;
 import org.apache.ambari.server.scheduler.ExecutionScheduleManager;
 import org.easymock.Capture;
 import org.junit.Assert;
 import org.junit.Test;
+import org.quartz.JobExecutionContext;
 
+import java.lang.reflect.Method;
 import java.util.HashMap;
 import java.util.Map;