You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by GitBox <gi...@apache.org> on 2020/06/01 04:10:35 UTC
[GitHub] [incubator-gobblin] arjun4084346 opened a new pull request #3011: [GOBBLIN-1160] No spec delete on gobblin service start
arjun4084346 opened a new pull request #3011:
URL: https://github.com/apache/incubator-gobblin/pull/3011
address review comments
improve test cases
code cleaning
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [x] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-1160
### Description
- [x] Here are some details about my PR, including screenshots (if applicable):
1) store specs in spec store only when compilation passes
2) fix the race condition between NonScheduledJobRunner and FlowCatalog threads
### Tests
- [x] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
added a few unit tests
### Commits
- [x] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] arjun4084346 closed pull request #3011: [GOBBLIN-1160] No spec delete on gobblin service start
Posted by GitBox <gi...@apache.org>.
arjun4084346 closed pull request #3011:
URL: https://github.com/apache/incubator-gobblin/pull/3011
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] asfgit closed pull request #3011: [GOBBLIN-1160] No spec delete on gobblin service start
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #3011:
URL: https://github.com/apache/incubator-gobblin/pull/3011
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] codecov-commenter commented on pull request #3011: [GOBBLIN-1160] No spec delete on gobblin service start
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #3011:
URL: https://github.com/apache/incubator-gobblin/pull/3011#issuecomment-636979327
# [Codecov](https://codecov.io/gh/apache/incubator-gobblin/pull/3011?src=pr&el=h1) Report
> :exclamation: No coverage uploaded for pull request base (`master@ee7e75a`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference#section-missing-base-commit).
> The diff coverage is `0.00%`.
[](https://codecov.io/gh/apache/incubator-gobblin/pull/3011?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #3011 +/- ##
========================================
Coverage ? 7.15%
Complexity ? 1247
========================================
Files ? 1945
Lines ? 73948
Branches ? 8187
========================================
Hits ? 5294
Misses ? 68147
Partials ? 507
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-gobblin/pull/3011?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [.../java/org/apache/gobblin/runtime/api/FlowSpec.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL0Zsb3dTcGVjLmphdmE=) | `25.73% <0.00%> (ø)` | `2.00 <0.00> (?)` | |
| [...ache/gobblin/runtime/spec\_catalog/FlowCatalog.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvc3BlY19jYXRhbG9nL0Zsb3dDYXRhbG9nLmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...in/service/modules/core/GobblinServiceManager.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9jb3JlL0dvYmJsaW5TZXJ2aWNlTWFuYWdlci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...bblin/service/modules/flow/MockedSpecCompiler.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9mbG93L01vY2tlZFNwZWNDb21waWxlci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../modules/scheduler/GobblinServiceJobScheduler.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9zY2hlZHVsZXIvR29iYmxpblNlcnZpY2VKb2JTY2hlZHVsZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...in/java/org/apache/gobblin/test/TimingManager.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi10ZXN0LXV0aWxzL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3Rlc3QvVGltaW5nTWFuYWdlci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00% <0.00%> (?%)` | |
| [...nagement/copy/replication/ReplicationMetaData.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L2NvcHkvcmVwbGljYXRpb24vUmVwbGljYXRpb25NZXRhRGF0YS5qYXZh) | `0.00% <0.00%> (ø)` | `0.00% <0.00%> (?%)` | |
| [.../gobblin/source/extractor/SimpleJsonExtractor.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NvdXJjZS9leHRyYWN0b3IvU2ltcGxlSnNvbkV4dHJhY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00% <0.00%> (?%)` | |
| [...he/gobblin/runtime/locks/JobLockEventListener.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvbG9ja3MvSm9iTG9ja0V2ZW50TGlzdGVuZXIuamF2YQ==) | `50.00% <0.00%> (ø)` | `1.00% <0.00%> (?%)` | |
| [...bblin/source/PartitionAwareFileRetrieverUtils.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree#diff-Z29iYmxpbi1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NvdXJjZS9QYXJ0aXRpb25Bd2FyZUZpbGVSZXRyaWV2ZXJVdGlscy5qYXZh) | `0.00% <0.00%> (ø)` | `0.00% <0.00%> (?%)` | |
| ... and [1940 more](https://codecov.io/gh/apache/incubator-gobblin/pull/3011/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-gobblin/pull/3011?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-gobblin/pull/3011?src=pr&el=footer). Last update [ee7e75a...d4c5c46](https://codecov.io/gh/apache/incubator-gobblin/pull/3011?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] arjun4084346 closed pull request #3011: [GOBBLIN-1160] No spec delete on gobblin service start
Posted by GitBox <gi...@apache.org>.
arjun4084346 closed pull request #3011:
URL: https://github.com/apache/incubator-gobblin/pull/3011
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] sv2000 commented on a change in pull request #3011: [GOBBLIN-1160] No spec delete on gobblin service start
Posted by GitBox <gi...@apache.org>.
sv2000 commented on a change in pull request #3011:
URL: https://github.com/apache/incubator-gobblin/pull/3011#discussion_r432755115
##########
File path: gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/service/GobblinServiceManagerTest.java
##########
@@ -180,50 +199,137 @@ public void cleanUp() throws Exception {
}
}
+ /**
+ * To test an existing flow in a spec store does not get deleted just because it is not compilable during service restarts
+ */
@Test
- public void testCreate() throws Exception {
- Map<String, String> flowProperties = Maps.newHashMap();
- flowProperties.put("param1", "value1");
- flowProperties.put(ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, TEST_SOURCE_NAME);
- flowProperties.put(ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, TEST_SINK_NAME);
+ public void testRestart() throws Exception {
+ FlowConfig flowConfig = new FlowConfig().setId(UNCOMPILABLE_FLOW_ID).setTemplateUris(TEST_TEMPLATE_URI)
+ .setSchedule(new Schedule().setCronSchedule(TEST_SCHEDULE).setRunImmediately(false))
+ .setProperties(new StringMap(flowProperties));
+ FlowSpec spec = FlowConfigResourceLocalHandler.createFlowSpecForConfig(flowConfig);
+ FlowConfig runOnceFlowConfig = new FlowConfig().setId(TEST_FLOW_ID)
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
+ FlowSpec runOnceSpec = FlowConfigResourceLocalHandler.createFlowSpecForConfig(runOnceFlowConfig);
+
+ // add the non compilable flow directly to the spec store skipping flow catalog which would not allow this
+ this.gobblinServiceManager.getFlowCatalog().getSpecStore().addSpec(spec);
+ this.gobblinServiceManager.getFlowCatalog().getSpecStore().addSpec(runOnceSpec);
+
+ List<Spec> specs = (List<Spec>) this.gobblinServiceManager.getFlowCatalog().getSpecs();
+
+ Assert.assertEquals(specs.size(), 2);
+ if (specs.get(0).getUri().equals(spec.getUri())) {
+ Assert.assertEquals(specs.get(1).getUri(), runOnceSpec.getUri());
+ } else if (specs.get(0).getUri().equals(runOnceSpec.getUri())) {
+ Assert.assertEquals(specs.get(1).getUri(), spec.getUri());
+ } else {
+ Assert.fail();
+ }
+
+ // restart the service
+ serviceReboot();
+
+ // runOnce job should get deleted from the spec store after running and uncompilable job should stay
+ AssertWithBackoff.create().maxSleepMs(200L).timeoutMs(20000L).backoffFactor(1)
+ .assertTrue(input -> this.gobblinServiceManager.getFlowCatalog().getSpecs().size() == 1,
+ "Waiting for the runOnce job to finish");
+
+ specs = (List<Spec>) this.gobblinServiceManager.getFlowCatalog().getSpecs();
+ Assert.assertEquals(specs.get(0).getUri(), spec.getUri());
+ Assert.assertFalse(flowConfig.getSchedule().isRunImmediately());
+
+ // clean it
+ this.gobblinServiceManager.getFlowCatalog().remove(spec.getUri());
+ specs = (List<Spec>) this.gobblinServiceManager.getFlowCatalog().getSpecs();
+ Assert.assertEquals(specs.size(), 0);
+ }
+
+ @Test (dependsOnMethods = "testRestart")
+ public void testUncompilableJob() throws Exception {
+ FlowId flowId = new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(MockedSpecCompiler.UNCOMPILABLE_FLOW);
+ URI uri = FlowConfigResourceLocalHandler.FlowUriUtils.createFlowSpecUri(flowId);
+ FlowConfig flowConfig = new FlowConfig().setId(flowId)
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
+ RestLiResponseException exception = null;
+ try {
+ this.flowConfigClient.createFlowConfig(flowConfig);
+ } catch (RestLiResponseException e) {
+ exception = e;
+ }
+ Assert.assertEquals(exception.getStatus(), HttpStatus.BAD_REQUEST_400);
+ // uncompilable job should not be persisted
+ Assert.assertEquals(this.gobblinServiceManager.getFlowCatalog().getSpecs().size(), 0);
+ Assert.assertFalse(this.gobblinServiceManager.getScheduler().getScheduledFlowSpecs().containsKey(uri.toString()));
+ }
+
+ @Test (dependsOnMethods = "testUncompilableJob")
+ public void testRunOnceJob() throws Exception {
+ FlowConfig flowConfig = new FlowConfig().setId(TEST_FLOW_ID)
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
+
+ this.flowConfigClient.createFlowConfig(flowConfig);
+
+ // runOnce job should not be persisted
+ AssertWithBackoff.create().maxSleepMs(200L).timeoutMs(2000L).backoffFactor(1)
+ .assertTrue(input -> this.gobblinServiceManager.getFlowCatalog().getSpecs().size() == 0,
+ "Flow that was created is not " + "reflecting in FlowCatalog");
Review comment:
Can we have a more descriptive message? I do not understand "not reflecting in FlowCatalog" really means.
##########
File path: gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/service/GobblinServiceManagerTest.java
##########
@@ -180,50 +199,137 @@ public void cleanUp() throws Exception {
}
}
+ /**
+ * To test an existing flow in a spec store does not get deleted just because it is not compilable during service restarts
+ */
@Test
- public void testCreate() throws Exception {
- Map<String, String> flowProperties = Maps.newHashMap();
- flowProperties.put("param1", "value1");
- flowProperties.put(ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, TEST_SOURCE_NAME);
- flowProperties.put(ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, TEST_SINK_NAME);
+ public void testRestart() throws Exception {
+ FlowConfig flowConfig = new FlowConfig().setId(UNCOMPILABLE_FLOW_ID).setTemplateUris(TEST_TEMPLATE_URI)
+ .setSchedule(new Schedule().setCronSchedule(TEST_SCHEDULE).setRunImmediately(false))
+ .setProperties(new StringMap(flowProperties));
+ FlowSpec spec = FlowConfigResourceLocalHandler.createFlowSpecForConfig(flowConfig);
+ FlowConfig runOnceFlowConfig = new FlowConfig().setId(TEST_FLOW_ID)
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
+ FlowSpec runOnceSpec = FlowConfigResourceLocalHandler.createFlowSpecForConfig(runOnceFlowConfig);
+
+ // add the non compilable flow directly to the spec store skipping flow catalog which would not allow this
+ this.gobblinServiceManager.getFlowCatalog().getSpecStore().addSpec(spec);
+ this.gobblinServiceManager.getFlowCatalog().getSpecStore().addSpec(runOnceSpec);
+
+ List<Spec> specs = (List<Spec>) this.gobblinServiceManager.getFlowCatalog().getSpecs();
+
+ Assert.assertEquals(specs.size(), 2);
+ if (specs.get(0).getUri().equals(spec.getUri())) {
+ Assert.assertEquals(specs.get(1).getUri(), runOnceSpec.getUri());
+ } else if (specs.get(0).getUri().equals(runOnceSpec.getUri())) {
+ Assert.assertEquals(specs.get(1).getUri(), spec.getUri());
+ } else {
+ Assert.fail();
+ }
+
+ // restart the service
+ serviceReboot();
+
+ // runOnce job should get deleted from the spec store after running and uncompilable job should stay
+ AssertWithBackoff.create().maxSleepMs(200L).timeoutMs(20000L).backoffFactor(1)
+ .assertTrue(input -> this.gobblinServiceManager.getFlowCatalog().getSpecs().size() == 1,
+ "Waiting for the runOnce job to finish");
+
+ specs = (List<Spec>) this.gobblinServiceManager.getFlowCatalog().getSpecs();
+ Assert.assertEquals(specs.get(0).getUri(), spec.getUri());
+ Assert.assertFalse(flowConfig.getSchedule().isRunImmediately());
+
+ // clean it
+ this.gobblinServiceManager.getFlowCatalog().remove(spec.getUri());
+ specs = (List<Spec>) this.gobblinServiceManager.getFlowCatalog().getSpecs();
+ Assert.assertEquals(specs.size(), 0);
+ }
+
+ @Test (dependsOnMethods = "testRestart")
+ public void testUncompilableJob() throws Exception {
+ FlowId flowId = new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(MockedSpecCompiler.UNCOMPILABLE_FLOW);
+ URI uri = FlowConfigResourceLocalHandler.FlowUriUtils.createFlowSpecUri(flowId);
+ FlowConfig flowConfig = new FlowConfig().setId(flowId)
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
+ RestLiResponseException exception = null;
+ try {
+ this.flowConfigClient.createFlowConfig(flowConfig);
+ } catch (RestLiResponseException e) {
+ exception = e;
+ }
+ Assert.assertEquals(exception.getStatus(), HttpStatus.BAD_REQUEST_400);
+ // uncompilable job should not be persisted
+ Assert.assertEquals(this.gobblinServiceManager.getFlowCatalog().getSpecs().size(), 0);
+ Assert.assertFalse(this.gobblinServiceManager.getScheduler().getScheduledFlowSpecs().containsKey(uri.toString()));
+ }
+
+ @Test (dependsOnMethods = "testUncompilableJob")
+ public void testRunOnceJob() throws Exception {
+ FlowConfig flowConfig = new FlowConfig().setId(TEST_FLOW_ID)
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
+
+ this.flowConfigClient.createFlowConfig(flowConfig);
+
+ // runOnce job should not be persisted
Review comment:
This comment is confusing. What you want to say here is that the runOnce job is deleted soon after it is run.
##########
File path: gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/service/GobblinServiceManagerTest.java
##########
@@ -180,50 +199,137 @@ public void cleanUp() throws Exception {
}
}
+ /**
+ * To test an existing flow in a spec store does not get deleted just because it is not compilable during service restarts
+ */
@Test
- public void testCreate() throws Exception {
- Map<String, String> flowProperties = Maps.newHashMap();
- flowProperties.put("param1", "value1");
- flowProperties.put(ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, TEST_SOURCE_NAME);
- flowProperties.put(ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, TEST_SINK_NAME);
+ public void testRestart() throws Exception {
+ FlowConfig flowConfig = new FlowConfig().setId(UNCOMPILABLE_FLOW_ID).setTemplateUris(TEST_TEMPLATE_URI)
+ .setSchedule(new Schedule().setCronSchedule(TEST_SCHEDULE).setRunImmediately(false))
+ .setProperties(new StringMap(flowProperties));
+ FlowSpec spec = FlowConfigResourceLocalHandler.createFlowSpecForConfig(flowConfig);
+ FlowConfig runOnceFlowConfig = new FlowConfig().setId(TEST_FLOW_ID)
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
+ FlowSpec runOnceSpec = FlowConfigResourceLocalHandler.createFlowSpecForConfig(runOnceFlowConfig);
+
+ // add the non compilable flow directly to the spec store skipping flow catalog which would not allow this
+ this.gobblinServiceManager.getFlowCatalog().getSpecStore().addSpec(spec);
+ this.gobblinServiceManager.getFlowCatalog().getSpecStore().addSpec(runOnceSpec);
+
+ List<Spec> specs = (List<Spec>) this.gobblinServiceManager.getFlowCatalog().getSpecs();
+
+ Assert.assertEquals(specs.size(), 2);
+ if (specs.get(0).getUri().equals(spec.getUri())) {
+ Assert.assertEquals(specs.get(1).getUri(), runOnceSpec.getUri());
+ } else if (specs.get(0).getUri().equals(runOnceSpec.getUri())) {
+ Assert.assertEquals(specs.get(1).getUri(), spec.getUri());
+ } else {
+ Assert.fail();
+ }
+
+ // restart the service
+ serviceReboot();
+
+ // runOnce job should get deleted from the spec store after running and uncompilable job should stay
+ AssertWithBackoff.create().maxSleepMs(200L).timeoutMs(20000L).backoffFactor(1)
+ .assertTrue(input -> this.gobblinServiceManager.getFlowCatalog().getSpecs().size() == 1,
+ "Waiting for the runOnce job to finish");
+
+ specs = (List<Spec>) this.gobblinServiceManager.getFlowCatalog().getSpecs();
+ Assert.assertEquals(specs.get(0).getUri(), spec.getUri());
+ Assert.assertFalse(flowConfig.getSchedule().isRunImmediately());
+
+ // clean it
+ this.gobblinServiceManager.getFlowCatalog().remove(spec.getUri());
+ specs = (List<Spec>) this.gobblinServiceManager.getFlowCatalog().getSpecs();
+ Assert.assertEquals(specs.size(), 0);
+ }
+
+ @Test (dependsOnMethods = "testRestart")
+ public void testUncompilableJob() throws Exception {
+ FlowId flowId = new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(MockedSpecCompiler.UNCOMPILABLE_FLOW);
+ URI uri = FlowConfigResourceLocalHandler.FlowUriUtils.createFlowSpecUri(flowId);
+ FlowConfig flowConfig = new FlowConfig().setId(flowId)
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
+ RestLiResponseException exception = null;
+ try {
+ this.flowConfigClient.createFlowConfig(flowConfig);
+ } catch (RestLiResponseException e) {
+ exception = e;
+ }
+ Assert.assertEquals(exception.getStatus(), HttpStatus.BAD_REQUEST_400);
+ // uncompilable job should not be persisted
+ Assert.assertEquals(this.gobblinServiceManager.getFlowCatalog().getSpecs().size(), 0);
+ Assert.assertFalse(this.gobblinServiceManager.getScheduler().getScheduledFlowSpecs().containsKey(uri.toString()));
+ }
+
+ @Test (dependsOnMethods = "testUncompilableJob")
+ public void testRunOnceJob() throws Exception {
+ FlowConfig flowConfig = new FlowConfig().setId(TEST_FLOW_ID)
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
+
+ this.flowConfigClient.createFlowConfig(flowConfig);
+
+ // runOnce job should not be persisted
+ AssertWithBackoff.create().maxSleepMs(200L).timeoutMs(2000L).backoffFactor(1)
+ .assertTrue(input -> this.gobblinServiceManager.getFlowCatalog().getSpecs().size() == 0,
+ "Flow that was created is not " + "reflecting in FlowCatalog");
+ AssertWithBackoff.create().maxSleepMs(100L).timeoutMs(1000L).backoffFactor(1)
+ .assertTrue(input -> !this.gobblinServiceManager.getScheduler().getScheduledFlowSpecs().containsKey(TEST_URI.toString()),
+ "Waiting for job to get orchestrated...");
+ }
+
+ @Test (dependsOnMethods = "testRunOnceJob")
+ public void testExplainJob() throws Exception {
FlowConfig flowConfig = new FlowConfig().setId(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME))
+ .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties)).setExplain(true);
+
+ this.flowConfigClient.createFlowConfig(flowConfig);
+
+ // explain job should not be persisted
+ Assert.assertEquals(this.gobblinServiceManager.getFlowCatalog().getSpecs().size(), 0,
+ "Flow that was created is not " + "reflecting in FlowCatalog");
+ Assert.assertFalse(this.gobblinServiceManager.getScheduler().getScheduledFlowSpecs().containsKey(TEST_URI.toString()));
+ }
+
+ @Test (dependsOnMethods = "testExplainJob")
+ public void testCreate() throws Exception {
+ FlowConfig flowConfig = new FlowConfig().setId(TEST_FLOW_ID)
.setTemplateUris(TEST_TEMPLATE_URI).setSchedule(new Schedule().setCronSchedule(TEST_SCHEDULE).setRunImmediately(true))
.setProperties(new StringMap(flowProperties));
this.flowConfigClient.createFlowConfig(flowConfig);
- Assert.assertTrue(this.gobblinServiceManager.getFlowCatalog().getSpecs().size() == 1, "Flow that was created is not "
- + "reflecting in FlowCatalog");
+ Assert.assertEquals(this.gobblinServiceManager.getFlowCatalog().getSpecs().size(), 1,
+ "Flow that was created is not " + "reflecting in FlowCatalog");
Review comment:
Same comment as earlier about the assertion failure message.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] arjun4084346 closed pull request #3011: [GOBBLIN-1160] No spec delete on gobblin service start
Posted by GitBox <gi...@apache.org>.
arjun4084346 closed pull request #3011:
URL: https://github.com/apache/incubator-gobblin/pull/3011
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] sv2000 commented on a change in pull request #3011: [GOBBLIN-1160] No spec delete on gobblin service start
Posted by GitBox <gi...@apache.org>.
sv2000 commented on a change in pull request #3011:
URL: https://github.com/apache/incubator-gobblin/pull/3011#discussion_r432781803
##########
File path: gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
##########
@@ -435,10 +454,17 @@ public void run() {
try {
GobblinServiceJobScheduler.this.runJob(this.jobConfig, this.jobListener);
if (flowCatalog.isPresent() && removeSpec) {
+ while (!GobblinServiceJobScheduler.this.flowCatalog.get().exists(specUri)) {
Review comment:
Should this block be replaced with a "wait with a timeout"? Is there a chance this thread may wait forever? If so, we should avoid that.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] sv2000 commented on a change in pull request #3011: [GOBBLIN-1160] No spec delete on gobblin service start
Posted by GitBox <gi...@apache.org>.
sv2000 commented on a change in pull request #3011:
URL: https://github.com/apache/incubator-gobblin/pull/3011#discussion_r432873482
##########
File path: gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/service/GobblinServiceManagerTest.java
##########
@@ -147,8 +165,8 @@ public void setup() throws Exception {
ConfigUtils.propertiesToConfig(serviceCoreProperties), Optional.of(new Path(SERVICE_WORK_DIR)));
this.gobblinServiceManager.start();
- this.flowConfigClient = new FlowConfigClient(String.format("http://localhost:%s/",
- this.gobblinServiceManager.getRestLiServer().getPort()));
+ this.flowConfigClient = new FlowConfigV2Client(String.format("http://localhost:%s/",
Review comment:
Can we replace localhost with 127.0.0.1 to avoid flakiness on Travis?
##########
File path: gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/FlowCatalog.java
##########
@@ -289,6 +297,9 @@ public Spec getSpecWrapper(URI uri) {
/**
* Persist {@link Spec} into {@link SpecStore} and notify {@link SpecCatalogListener} if triggerListener
* is set to true.
+ * If the {@link Spec} is a {@link FlowSpec} it is persisted if it can be compiled at the time this method received
+ * the spec. `explain` specs are not persisted. The logic of this method is tightly coupled with the logic of
Review comment:
Javadoc can be enhanced to explain how the coupling is achieved e.g. we use condition variables to achieve synchronization between NonScheduledJobRunner thread and FlowCatalog thread to ensure deletion of FlowSpec happens after the corresponding run once flow is submitted to the orchestrator.
##########
File path: gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
##########
@@ -435,10 +452,22 @@ public void run() {
try {
GobblinServiceJobScheduler.this.runJob(this.jobConfig, this.jobListener);
if (flowCatalog.isPresent() && removeSpec) {
+ if (GobblinServiceJobScheduler.this.flowCatalog.get().getSpecSyncObjects().containsKey(specUri.toString())) {
+ // if the sync object does not exist, this job must be set to run due to job submission at service restart
+ Object syncObject = GobblinServiceJobScheduler.this.flowCatalog.get().getSpecSyncObjects().get(specUri.toString());
Review comment:
This looks very cumbersome! Let's add a method in flowCatalog that returns the syncObject given the specUri and avoid exposing specSyncObjects map outside flowcatalog.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org