allow handle persistor in sync waypoint (LiveRamp#12)

Author: Charles Giardina

src/java/com/liveramp/captain/waypoint/SyncWaypoint.java

@@ -5,10 +5,12 @@
 import com.liveramp.captain.config_retriever.CaptainConfigRetriever;
 import com.liveramp.captain.config_retriever.CaptainConfigRetrieverFactory;
 import com.liveramp.captain.config_retriever.CaptainConfigRetrieverWrapperFactory;
+import com.liveramp.captain.handle_persistor.CaptainHandlePersistor;
+import com.liveramp.captain.handle_persistor.CaptainHandlePersistorFactory;
+import com.liveramp.captain.handle_persistor.CaptainHandlePersistorWrapperFactory;
 import com.liveramp.captain.request_submitter.CaptainRequestSubmitter;
 import com.liveramp.captain.request_submitter.CaptainRequestSubmitterFactory;
 import com.liveramp.captain.request_submitter.CaptainRequestSubmitterWrapperFactory;
-import com.liveramp.captain.status_retriever.CaptainStatusRetriever;
 import com.liveramp.captain.status_retriever.CaptainStatusRetrieverFactory;
 import com.liveramp.cl_types.clr.ClStep;
 
@@ -19,19 +21,36 @@ public class SyncWaypoint<T extends Serializable, ServiceHandle> implements Wayp
   private SyncWaypoint(
       ClStep step,
       CaptainConfigRetrieverFactory<T> configRetriever,
-      CaptainRequestSubmitterFactory<T, ServiceHandle> requestSubmitter) {
+      CaptainRequestSubmitterFactory<T, ServiceHandle> requestSubmitter,
+      CaptainHandlePersistorFactory<ServiceHandle> handlePersistor) {
     this.step = step;
-    this.waypointSubmitter = new WaypointSubmitterImpl<>(step, configRetriever, requestSubmitter, null);
+    this.waypointSubmitter = new WaypointSubmitterImpl<>(step, configRetriever, requestSubmitter, handlePersistor);
+  }
+
+  private SyncWaypoint(
+      ClStep step,
+      CaptainConfigRetrieverFactory<T> configRetriever,
+      CaptainRequestSubmitterFactory<T, ServiceHandle> requestSubmitter) {
+    this(step, configRetriever, requestSubmitter, null);
+  }
+
+  public static <T extends Serializable, ServiceHandle> SyncWaypoint<T, ServiceHandle> of(ClStep step, CaptainConfigRetrieverFactory<T> configRetriever, CaptainRequestSubmitterFactory<T, ServiceHandle> requestSubmitter, CaptainHandlePersistorFactory<ServiceHandle> handlePersistor) {
+    return new SyncWaypoint<>(step, configRetriever, requestSubmitter, handlePersistor);
   }
 
   public static <T extends Serializable, ServiceHandle> SyncWaypoint<T, ServiceHandle> of(ClStep step, CaptainConfigRetrieverFactory<T> configRetriever, CaptainRequestSubmitterFactory<T, ServiceHandle> requestSubmitter) {
-    return new SyncWaypoint<T, ServiceHandle>(step, configRetriever, requestSubmitter);
+    return new SyncWaypoint<>(step, configRetriever, requestSubmitter);
+  }
+
+  public static <T extends Serializable, ServiceHandle> SyncWaypoint<T, ServiceHandle> of(ClStep step, CaptainConfigRetriever<T> configRetriever, CaptainRequestSubmitter<T, ServiceHandle> requestSubmitter, CaptainHandlePersistor<ServiceHandle> handlePersistor) {
+    return of(step, new CaptainConfigRetrieverWrapperFactory<T>(configRetriever), new CaptainRequestSubmitterWrapperFactory<T, ServiceHandle>(requestSubmitter), new CaptainHandlePersistorWrapperFactory<ServiceHandle>(handlePersistor));
   }
 
   public static <T extends Serializable, ServiceHandle> SyncWaypoint<T, ServiceHandle> of(ClStep step, CaptainConfigRetriever<T> configRetriever, CaptainRequestSubmitter<T, ServiceHandle> requestSubmitter) {
     return of(step, new CaptainConfigRetrieverWrapperFactory<T>(configRetriever), new CaptainRequestSubmitterWrapperFactory<T, ServiceHandle>(requestSubmitter));
   }
 
+
   @Override
   public ClStep getStep() {
     return step;

test/java/com/liveramp/captain/daemon/TestClCaptainJoblet.java

@@ -76,6 +76,7 @@ public void setup() {
 
     testManifest = new DefaultManifestImpl(Lists.newArrayList(
         SyncWaypoint.of(ClStep.COLUMN_CREATION, configRetrieverStep1, requestSubmitterStep1),
+        SyncWaypoint.of(ClStep.CLA_VALIDATION, configRetrieverStep2, requestSubmitterStep2, handlePersistor2),
         AsyncWaypoint.of(ClStep.PRIVACY_INQUISITOR, configRetrieverStep2, requestSubmitterStep2, handlePersistor2, captainStatusRetrieverStep2),
         FlowControlWaypoint.of(ClStep.PRIVACY_INQUISITOR_SENTENCING, captainStatusRetrieverStep3),
         AsyncWaypoint.of(ClStep.HEARTBREAKER, configRetrieverStep4, requestSubmitterStep4, handlePersistor4, captainStatusRetrieverStep4),
@@ -95,7 +96,7 @@ public void testGoToNextStep() throws DaemonException, TException {
 
     new CaptainJoblet(config, alertsHandler, requestUpdater, manifestMap, true).run();
 
-    verify(requestUpdater, times(1)).setStepAndStatus(config.getJobId(), ClStep.COLUMN_CREATION, CaptStatus.COMPLETED, ClStep.PRIVACY_INQUISITOR, CaptStatus.READY);
+    verify(requestUpdater, times(1)).setStepAndStatus(config.getJobId(), ClStep.COLUMN_CREATION, CaptStatus.COMPLETED, ClStep.CLA_VALIDATION, CaptStatus.READY);
   }
 
   @Test
@@ -162,6 +163,22 @@ public void testSubmitRequestSync() throws TException, DaemonException {
     verify(requestUpdater, times(1)).setStatus(config.getJobId(), ClStep.COLUMN_CREATION, CaptStatus.READY, CaptStatus.COMPLETED);
   }
 
+  @Test
+  public void testSubmitRequestSyncWithHandlerPersistor() throws TException, DaemonException {
+    CaptainRequestConfig config = new CaptainRequestConfig(JOB_ID, CaptStatus.READY, ClStep.CLA_VALIDATION, ServiceRequestPriority.DEFAULT, ClaType.OFFLINE_INDIVIDUAL);
+
+    PInquisitorRequestConfig mockPinqRequest = mock(PInquisitorRequestConfig.class);
+    when(configRetrieverStep2.retrieveConfig(JOB_ID)).thenReturn(mockPinqRequest);
+    when(requestSubmitterStep2.submit(mockPinqRequest, new RequestOptions())).thenReturn(SERVICE_HANDLE);
+
+    new CaptainJoblet(config, alertsHandler, requestUpdater, manifestMap, true).run();
+
+    verify(configRetrieverStep2, times(1)).retrieveConfig(config.getJobId());
+    verify(requestSubmitterStep2, times(1)).submit(mockPinqRequest, new RequestOptions());
+    verify(handlePersistor2, times(1)).persist(JOB_ID, SERVICE_HANDLE, ClStep.CLA_VALIDATION);
+    verify(requestUpdater, times(1)).setStatus(config.getJobId(), ClStep.CLA_VALIDATION, CaptStatus.READY, CaptStatus.COMPLETED);
+  }
+
   @Test
   public void testSubmitRequestFlowControl1() throws TException, DaemonException {
     CaptainRequestConfig config = new CaptainRequestConfig(JOB_ID, CaptStatus.READY, ClStep.PRIVACY_INQUISITOR_SENTENCING, ServiceRequestPriority.DEFAULT, ClaType.OFFLINE_INDIVIDUAL);