[GOBBLIN-2136] consolidate various resource handlers for flowconfigs

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/idl/org.apache.gobblin.service.flowstatuses.restspec.json

@@ -6,7 +6,7 @@
   "namespace" : "org.apache.gobblin.service",
   "path" : "/flowstatuses",
   "schema" : "org.apache.gobblin.service.FlowStatus",
-  "doc" : "Resource for handling flow status requests. Deprecated, use {@link FlowExecutionResource}\n\ngenerated from: org.apache.gobblin.service.FlowStatusResource",
+  "doc" : "Deprecated, use {@link FlowExecutionResource}\n\nResource for handling flow status requests. generated from: org.apache.gobblin.service.FlowStatusResource",
   "collection" : {
     "identifier" : {
       "name" : "id",

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json

@@ -271,7 +271,7 @@
     "namespace" : "org.apache.gobblin.service",
     "path" : "/flowstatuses",
     "schema" : "org.apache.gobblin.service.FlowStatus",
-    "doc" : "Resource for handling flow status requests. Deprecated, use {@link FlowExecutionResource}\n\ngenerated from: org.apache.gobblin.service.FlowStatusResource",
+    "doc" : "Deprecated, use {@link FlowExecutionResource}\n\nResource for handling flow status requests. generated from: org.apache.gobblin.service.FlowStatusResource",
     "collection" : {
       "identifier" : {
         "name" : "id",

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowConfigsResourceHandlerInterface.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service;
+
+import java.util.Collection;
+import java.util.Properties;
+
+import com.linkedin.restli.common.ComplexResourceKey;
+import com.linkedin.restli.common.PatchRequest;
+import com.linkedin.restli.server.CreateKVResponse;
+import com.linkedin.restli.server.UpdateResponse;
+
+import org.apache.gobblin.runtime.api.FlowSpecSearchObject;
+
+
+// This is an interface rather than a class because implementation may need resources from the packages it cannot have
+// direct dependency on
+public interface FlowConfigsResourceHandlerInterface {
+
+  FlowConfig getFlowConfig(FlowId flowId) throws FlowConfigLoggedException;
+
+  Collection<FlowConfig> getFlowConfig(FlowSpecSearchObject flowSpecSearchObject) throws FlowConfigLoggedException;
+
+  /**
+   * Get all flow configs
+   */
+  Collection<FlowConfig> getAllFlowConfigs();
+
+  /**
+   * Get all flow configs in between start and start + count - 1
+   */
+  Collection<FlowConfig> getAllFlowConfigs(int start, int count);
+
+  UpdateResponse deleteFlowConfig(FlowId flowId, Properties header) throws FlowConfigLoggedException;
+
+  UpdateResponse  partialUpdateFlowConfig(FlowId flowId, PatchRequest<FlowConfig> flowConfigPatch) throws FlowConfigLoggedException;
+
+  UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig) throws FlowConfigLoggedException;
+
+  CreateKVResponse<ComplexResourceKey<FlowId, FlowStatusId>, FlowConfig> createFlowConfig(FlowConfig flowConfig) throws FlowConfigLoggedException;
+}

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResourceHandlerInterface.java

@@ -25,8 +25,8 @@
 import com.linkedin.restli.server.UpdateResponse;
 
 
-// Unlike FlowConfigsV2ResourceHandler, this is an interface rather than a class because it's implementation needs
-// classes from gobblin-service module, and adding gobblin-service as a dependency will cause circular dependency,
+// This is an interface rather than a class because implementation may need resources from the packages it cannot have
+// direct dependency on
 public interface FlowExecutionResourceHandlerInterface {
   /**
    * Get {@link FlowExecution}

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/modules/restli/FlowConfigsV2ResourceHandler.java

@@ -66,6 +66,7 @@
 import org.apache.gobblin.runtime.util.InjectionNames;
 import org.apache.gobblin.service.FlowConfig;
 import org.apache.gobblin.service.FlowConfigLoggedException;
+import org.apache.gobblin.service.FlowConfigsResourceHandlerInterface;
 import org.apache.gobblin.service.FlowId;
 import org.apache.gobblin.service.FlowStatusId;
 import org.apache.gobblin.service.RequesterService;
@@ -75,7 +76,7 @@
 
 
 @Slf4j
-public class FlowConfigsV2ResourceHandler {
+public class FlowConfigsV2ResourceHandler implements FlowConfigsResourceHandlerInterface {
 
   @Getter
   private String serviceName;
@@ -164,16 +165,14 @@ public UpdateResponse  partialUpdateFlowConfig(FlowId flowId,
     return updateFlowConfig(flowId, flowConfig, modifiedWatermark);
   }
 
-  public UpdateResponse updateFlowConfig(FlowId flowId,
-      FlowConfig flowConfig) throws FlowConfigLoggedException {
+  public UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig) throws FlowConfigLoggedException {
     // We have modifiedWatermark here to avoid update config happens at the same time on different hosts overwrite each other
     // timestamp here will be treated as largest modifiedWatermark that we can update
     long version = System.currentTimeMillis() / 1000;
     return updateFlowConfig(flowId, flowConfig, version);
   }
 
-  public UpdateResponse updateFlowConfig(FlowId flowId,
-      FlowConfig flowConfig, long modifiedWatermark) throws FlowConfigLoggedException {
+  private UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig, long modifiedWatermark) throws FlowConfigLoggedException {
     String flowName = flowId.getFlowName();
     String flowGroup = flowId.getFlowGroup();
 

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/test/java/org/apache/gobblin/service/FlowConfigResourceLocalHandlerTest.java → gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/test/java/org/apache/gobblin/service/FlowConfigsV2ResourceHandlerTest.java

@@ -32,7 +32,7 @@
 import org.apache.gobblin.service.modules.restli.FlowConfigsV2ResourceHandler;
 
 
-public class FlowConfigResourceLocalHandlerTest {
+public class FlowConfigsV2ResourceHandlerTest {
   private static final String TEST_GROUP_NAME = "testGroup1";
   private static final String TEST_FLOW_NAME = "testFlow1";
   private static final String TEST_SCHEDULE = "0 1/0 * ? * *";

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/test/java/org/apache/gobblin/service/FlowExecutionResourceLocalHandlerTest.java → gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/test/java/org/apache/gobblin/service/FlowExecutionResourceHandlerTest.java

@@ -21,7 +21,7 @@
 import org.testng.annotations.Test;
 
 
-public class FlowExecutionResourceLocalHandlerTest {
+public class FlowExecutionResourceHandlerTest {
 
   @Test
   public void testEstimateCopyTimeLeftSanityCheck() {

gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceManager.java

@@ -280,7 +280,7 @@ private void registerServicesInLauncher(){
   private void configureServices(){
     if (configuration.isRestLIServerEnabled()) {
       this.restliServer = EmbeddedRestliServer.builder()
-          .resources(Lists.newArrayList(FlowConfigsV2Resource.class, FlowConfigsV2Resource.class))
+          .resources(Lists.newArrayList(FlowConfigsV2Resource.class))
           .injector(injector)
           .build();
 

gobblin-service/src/main/java/org/apache/gobblin/service/modules/restli/FlowExecutionResourceHandler.java

@@ -45,6 +45,9 @@
 import org.apache.gobblin.service.monitoring.FlowStatusGenerator;
 
 
+/**
+ * This is associated with {@link FlowExecutionResource} and handles all the requests FlowExecutionResource get.
+ */
 @Slf4j
 public class FlowExecutionResourceHandler implements FlowExecutionResourceHandlerInterface {
   private final DagManagementStateStore dagManagementStateStore;

gobblin-service/src/test/java/org/apache/gobblin/service/FlowExecutionTest.java → gobblin-service/src/test/java/org/apache/gobblin/service/FlowExecutionClientTest.java

@@ -46,7 +46,7 @@
 
 
 @Test(groups = { "gobblin.service" }, singleThreaded = true)
-public class FlowExecutionTest {
+public class FlowExecutionClientTest {
   private FlowExecutionClient client;
   private EmbeddedRestliServer _server;
   private List<List<org.apache.gobblin.service.monitoring.JobStatus>> _listOfJobStatusLists;

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

@@ -63,6 +63,25 @@ public class DagUtilsTest {
       .withValue(ConfigurationKeys.SPECEXECUTOR_INSTANCE_URI_KEY, ConfigValueFactory.fromAnyRef(
           MySqlDagManagementStateStoreTest.TEST_SPEC_EXECUTOR_URI));
 
+  @Test
+  void slaConfigCheck() throws Exception {
+    Dag<JobExecutionPlan> dag = DagTestUtils.buildDag("5", 123456783L, "FINISH_RUNNING", 1);
+    Assert.assertEquals(DagUtils.getFlowFinishDeadline(dag.getStartNodes().get(0)), ServiceConfigKeys.DEFAULT_FLOW_FINISH_DEADLINE_MILLIS);
+
+    Config jobConfig = dag.getStartNodes().get(0).getValue().getJobSpec().getConfig();
+    jobConfig = jobConfig
+        .withValue(ConfigurationKeys.GOBBLIN_FLOW_FINISH_DEADLINE_TIME, ConfigValueFactory.fromAnyRef("7"))
+        .withValue(ConfigurationKeys.GOBBLIN_FLOW_FINISH_DEADLINE_TIME_UNIT, ConfigValueFactory.fromAnyRef(TimeUnit.SECONDS.name()));
+    dag.getStartNodes().get(0).getValue().getJobSpec().setConfig(jobConfig);
+    Assert.assertEquals(DagUtils.getFlowFinishDeadline(dag.getStartNodes().get(0)), TimeUnit.SECONDS.toMillis(7L));
+
+    Config jobConfig2 = jobConfig
+        .withValue(ConfigurationKeys.GOBBLIN_FLOW_FINISH_DEADLINE_TIME, ConfigValueFactory.fromAnyRef("8"))
+        .withValue(ConfigurationKeys.GOBBLIN_FLOW_FINISH_DEADLINE_TIME_UNIT, ConfigValueFactory.fromAnyRef(TimeUnit.MINUTES.name()));
+    dag.getStartNodes().get(0).getValue().getJobSpec().setConfig(jobConfig2);
+    Assert.assertEquals(DagUtils.getFlowFinishDeadline(dag.getStartNodes().get(0)), TimeUnit.MINUTES.toMillis(8L));
+  }
+
   @Test
   void deadlineConfigCheck() throws Exception {
     Dag<JobExecutionPlan> dag = DagTestUtils.buildDag("5", 123456783L, "FINISH_RUNNING", 1);