Update unit tests

Author: vsinghal85

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagementStateStore.java

@@ -108,7 +108,7 @@ public interface DagManagementStateStore {
    * @param leaseParams   uniquely identifies the flow, the present action upon it, the time the action
    *                      was triggered, and if the dag action event we're checking on is a reminder event
    */
-  boolean canAcquireLeaseOnEntity(DagActionStore.LeaseParams leaseParams) throws IOException;
+  boolean isLeaseAcquirable(DagActionStore.LeaseParams leaseParams) throws IOException;
 
   /**
    * Returns the requested {@link  org.apache.gobblin.service.modules.flowgraph.Dag.DagNode} and its {@link JobStatus}.

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/InstrumentedLeaseArbiter.java

@@ -91,8 +91,8 @@ public LeaseAttemptStatus tryAcquireLease(DagActionStore.LeaseParams leaseParams
   }
 
   @Override
-  public boolean canAcquireLeaseOnEntity(DagActionStore.LeaseParams leaseParams) throws IOException {
-    return decoratedMultiActiveLeaseArbiter.canAcquireLeaseOnEntity(leaseParams);
+  public boolean isLeaseAcquirable(DagActionStore.LeaseParams leaseParams) throws IOException {
+    return decoratedMultiActiveLeaseArbiter.isLeaseAcquirable(leaseParams);
   }
 
   @Override

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MultiActiveLeaseArbiter.java

@@ -64,11 +64,12 @@ LeaseAttemptStatus tryAcquireLease(DagActionStore.LeaseParams leaseParams, boole
   /**
    * This method checks if lease can be acquired on provided flow in lease params
    * returns true if entry for the same flow does not exists within epsilon time
-   * in leaseArbiterStore
+   * in leaseArbiterStore, else returns false
    * @param leaseParams   uniquely identifies the flow, the present action upon it, the time the action
    *                      was triggered, and if the dag action event we're checking on is a reminder event
+   * @return true if lease can be acquired on the flow passed in the lease params, false otherwise
    */
-  boolean canAcquireLeaseOnEntity(DagActionStore.LeaseParams leaseParams)
+  boolean isLeaseAcquirable(DagActionStore.LeaseParams leaseParams)
       throws IOException;
 
   /**

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStore.java

@@ -171,8 +171,8 @@ public synchronized void updateDagNode(Dag.DagNode<JobExecutionPlan> dagNode)
   }
 
   @Override
-  public boolean canAcquireLeaseOnEntity(DagActionStore.LeaseParams leaseParams) throws IOException {
-    return multiActiveLeaseArbiter.canAcquireLeaseOnEntity(leaseParams);
+  public boolean isLeaseAcquirable(DagActionStore.LeaseParams leaseParams) throws IOException {
+    return multiActiveLeaseArbiter.isLeaseAcquirable(leaseParams);
   }
 
   @Override

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/MysqlMultiActiveLeaseArbiter.java

@@ -362,8 +362,12 @@ else if (leaseValidityStatus == 2) {
     }
   }
 
+  /*
+    Determines if a lease can be acquired for the given flow. A lease is acquirable if
+    no existing lease record exists in arbiter table or the record is older then epsilon time
+   */
   @Override
-  public boolean canAcquireLeaseOnEntity(DagActionStore.LeaseParams leaseParams) throws IOException {
+  public boolean isLeaseAcquirable(DagActionStore.LeaseParams leaseParams) throws IOException {
     Optional<GetEventInfoResult> infoResult = getExistingEventInfo(leaseParams);
     return infoResult.isPresent() ? !infoResult.get().isWithinEpsilon() : true;
   }

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/Orchestrator.java

@@ -17,8 +17,6 @@
 
 package org.apache.gobblin.service.modules.orchestration;
 
-import com.linkedin.restli.common.HttpStatus;
-import com.linkedin.restli.server.RestLiServiceException;
 import java.io.IOException;
 import java.net.URI;
 import java.util.Collections;
@@ -139,16 +137,22 @@ public AddSpecResponse onAddSpec(Spec addedSpec) {
     return new AddSpecResponse<>(null);
   }
 
+  /*
+    validates if lease can be acquired on the provided flowSpec,
+    else throw LeaseUnavailableException
+   */
   private void validateAdhocFlowLeasability(FlowSpec flowSpec) {
     if (!flowSpec.isScheduled()) {
       Config flowConfig = flowSpec.getConfig();
       String flowGroup = flowConfig.getString(ConfigurationKeys.FLOW_GROUP_KEY);
       String flowName = flowConfig.getString(ConfigurationKeys.FLOW_NAME_KEY);
+
       DagActionStore.DagAction dagAction = DagActionStore.DagAction.forFlow(flowGroup, flowName,
           FlowUtils.getOrCreateFlowExecutionId(flowSpec), DagActionStore.DagActionType.LAUNCH);
       DagActionStore.LeaseParams leaseParams = new DagActionStore.LeaseParams(dagAction, System.currentTimeMillis());
+      _log.info("validation of lease acquirability of adhoc flow with lease params: " + leaseParams);
       try {
-        if (!dagManagementStateStore.canAcquireLeaseOnEntity(leaseParams)) {
+        if (!dagManagementStateStore.isLeaseAcquirable(leaseParams)) {
           throw new LeaseUnavailableException("Lease already occupied by another execution of this flow");
         }
       } catch (IOException exception) {

gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MySqlDagManagementStateStoreTest.java

@@ -97,12 +97,12 @@ public static <T> boolean compareLists(List<T> list1, List<T> list2) {
 
   @Test
   public void testcanAcquireLeaseOnEntity() throws Exception{
-    Mockito.when(leaseArbiter.canAcquireLeaseOnEntity(Mockito.any(DagActionStore.LeaseParams.class))).thenReturn(true);
+    Mockito.when(leaseArbiter.isLeaseAcquirable(Mockito.any(DagActionStore.LeaseParams.class))).thenReturn(true);
     String flowName = "testFlow";
     String flowGroup = "testGroup";
     DagActionStore.DagAction dagAction = new DagActionStore.DagAction(flowName, flowGroup, System.currentTimeMillis(), "testJob", DagActionStore.DagActionType.LAUNCH);
     DagActionStore.LeaseParams leaseParams = new DagActionStore.LeaseParams(dagAction);
-    Assert.assertTrue(dagManagementStateStore.canAcquireLeaseOnEntity(leaseParams));
+    Assert.assertTrue(dagManagementStateStore.isLeaseAcquirable(leaseParams));
   }
 
   @Test

gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/MysqlMultiActiveLeaseArbiterTest.java

@@ -223,7 +223,7 @@ public void testWhenLeasableEntityUnavailable() throws Exception{
     Assert.assertTrue(firstLaunchStatus instanceof LeaseAttemptStatus.LeaseObtainedStatus);
     completeLeaseHelper(launchLeaseParams3);
     Thread.sleep(LESS_THAN_EPSILON);
-    Assert.assertFalse(mysqlMultiActiveLeaseArbiter.canAcquireLeaseOnEntity(launchLeaseParams3));
+    Assert.assertFalse(mysqlMultiActiveLeaseArbiter.isLeaseAcquirable(launchLeaseParams3));
   }
 
   /*
@@ -236,7 +236,7 @@ public void testWhenLeasableEntityAvailable() throws Exception{
     Assert.assertTrue(firstLaunchStatus instanceof LeaseAttemptStatus.LeaseObtainedStatus);
     completeLeaseHelper(launchLeaseParams4);
     Thread.sleep(MORE_THAN_EPSILON);
-    Assert.assertTrue(mysqlMultiActiveLeaseArbiter.canAcquireLeaseOnEntity(launchLeaseParams4));
+    Assert.assertTrue(mysqlMultiActiveLeaseArbiter.isLeaseAcquirable(launchLeaseParams4));
   }
 
   /*

gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/OrchestratorTest.java

@@ -326,18 +326,40 @@ public void createFlowSpec() throws Throwable {
      and throw LeaseUnavailableException
    */
   @Test(expectedExceptions = LeaseUnavailableException.class)
-  public void testOnAddSpec_withFlowSpec_leaseUnavailable() throws IOException {
+  public void onAddSpecForAdhocFlowThrowLeaseUnavailable() throws IOException {
     ConfigBuilder configBuilder = ConfigBuilder.create()
         .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "testGroup")
         .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "testName");
     Config config = configBuilder.build();
     FlowSpec flowSpec = FlowSpec.builder().withConfig(config).build();
-    Mockito.when(dagManagementStateStore.canAcquireLeaseOnEntity(Mockito.any(DagActionStore.LeaseParams.class))).thenReturn(false);
+    Mockito.when(dagManagementStateStore.isLeaseAcquirable(Mockito.any(DagActionStore.LeaseParams.class))).thenReturn(false);
     dagMgrNotFlowLaunchHandlerBasedOrchestrator.onAddSpec(flowSpec);
   }
 
+  /*
+   If no other flow has acquired lease within the epsilon time, then flow
+   compilation and addition to the store occurs normally
+ */
+  @Test
+  public void onAddSpecForAdhocFlowLeaseAvailable() throws IOException {
+    ConfigBuilder configBuilder = ConfigBuilder.create()
+        .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "testGroup")
+        .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "testName")
+        .addPrimitive("gobblin.flow.sourceIdentifier", "source")
+        .addPrimitive("gobblin.flow.destinationIdentifier", "destination");
+    Config config = configBuilder.build();
+    FlowSpec flowSpec = FlowSpec.builder().withConfig(config).build();
+    Mockito.when(dagManagementStateStore.isLeaseAcquirable(Mockito.any(DagActionStore.LeaseParams.class))).thenReturn(true);
+    AddSpecResponse addSpecResponse = dagMgrNotFlowLaunchHandlerBasedOrchestrator.onAddSpec(flowSpec);
+    Assert.assertNotNull(addSpecResponse);
+  }
+
+  /*
+    For Scheduled flow lease acquirable check does not occur,
+    and flow compilation occurs successfully
+   */
   @Test
-  public void testOnAddSpec_withFlowSpec_Available() throws IOException {
+  public void onAddSpecForScheduledFlow() throws IOException {
     ConfigBuilder configBuilder = ConfigBuilder.create()
         .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "testGroup")
         .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "testName")
@@ -346,7 +368,6 @@ public void testOnAddSpec_withFlowSpec_Available() throws IOException {
         .addPrimitive("gobblin.flow.destinationIdentifier", "destination");
     Config config = configBuilder.build();
     FlowSpec flowSpec = FlowSpec.builder().withConfig(config).build();
-    Mockito.when(dagManagementStateStore.canAcquireLeaseOnEntity(Mockito.any(DagActionStore.LeaseParams.class))).thenReturn(true);
     AddSpecResponse response = new AddSpecResponse<>(new Object());
     Mockito.when(specCompiler.onAddSpec(flowSpec)).thenReturn(response);
     AddSpecResponse addSpecResponse = dagMgrNotFlowLaunchHandlerBasedOrchestrator.onAddSpec(flowSpec);