address review comments

apache · Sep 9, 2024 · cf2d4e8 · cf2d4e8
1 parent 295b529
commit cf2d4e8
Show file tree

Hide file tree

Showing 10 changed files with 29 additions and 65 deletions.
diff --git a/...low-config-service-api/src/main/idl/org.apache.gobblin.service.flowstatuses.restspec.json b/...low-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",

diff --git a/...onfig-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json b/...onfig-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",

diff --git a/...erver/src/main/java/org/apache/gobblin/service/FlowExecutionResourceHandlerInterface.java b/...erver/src/main/java/org/apache/gobblin/service/FlowExecutionResourceHandlerInterface.java
@@ -26,7 +26,7 @@
 
 
 // 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,
+// classes from gobblin-service module, and adding gobblin-service as a dependency will cause circular dependency
 public interface FlowExecutionResourceHandlerInterface {
   /**
    * Get {@link FlowExecution}

diff --git a/...e/FlowConfigResourceLocalHandlerTest.java → ...ice/FlowConfigsV2ResourceHandlerTest.java b/...e/FlowConfigResourceLocalHandlerTest.java → ...ice/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 * ? * *";

diff --git a/...lowExecutionResourceLocalHandlerTest.java → ...ice/FlowExecutionResourceHandlerTest.java b/...lowExecutionResourceLocalHandlerTest.java → ...ice/FlowExecutionResourceHandlerTest.java
@@ -21,7 +21,7 @@
 import org.testng.annotations.Test;
 
 
-public class FlowExecutionResourceLocalHandlerTest {
+public class FlowExecutionResourceHandlerTest {
 
   @Test
   public void testEstimateCopyTimeLeftSanityCheck() {

diff --git a/...-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceManager.java b/...-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();
 

diff --git a/...src/main/java/org/apache/gobblin/service/modules/restli/FlowExecutionResourceHandler.java b/...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;

diff --git a/...he/gobblin/service/FlowExecutionTest.java → ...blin/service/FlowExecutionClientTest.java b/...he/gobblin/service/FlowExecutionTest.java → ...blin/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;

diff --git a/...n-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagFlowTest.java b/...n-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagFlowTest.java
diff --git a/...-service/src/test/java/org/apache/gobblin/service/modules/orchestration/DagUtilsTest.java b/...-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);