[GOBBLIN-2056] initialize topology specs directly without waitging for listener call… #3937

arjun4084346 · 2024-04-29T09:33:47Z

…backs

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

My PR addresses the following Gobblin JIRA issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-2056

Description

Here are some details about my PR, including screenshots (if applicable):
topologies are static and fixed. they can be initialized through configs. but right now we initialize them in spec compiler making spec compiler as a topology spec catalog listener and then wait for topology spec catalog to populate topologies.
this sometimes take longer time because it happens when some of the services' setActive is called. some newly created classes like DagManagementStateStore initialization may fail if topologies are not present by that time.
so in this pr i am populating topologies sooner in spec compiler

Tests

My PR adds the following unit tests OR does not need testing for this extremely good reason:
updated existing test cases

Commits

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":
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"

…backs

Will-Lo

Don't you need to load the topology spec store earlier in the GobblinServiceManager? Otherwise the topology map being passed to the flow compilation validation helper can be empty right?

Will-Lo · 2024-04-29T15:10:54Z

...service/src/main/java/org/apache/gobblin/service/modules/flow/BaseFlowToJobSpecCompiler.java

-
-  public BaseFlowToJobSpecCompiler(Config config, Optional<Logger> log, boolean instrumentationEnabled){
-    this.log = log.isPresent() ? log.get() : LoggerFactory.getLogger(getClass());
-    if (instrumentationEnabled) {


Is the intention here to remove the use of this variable for checking if instrumentation is enabled? It's still used widely in GaaS modules so there should be justification

Arjun did some refactoring before which made instrumentation required I believe. Can you link that change to explain what's happening here?

yes, in https://github.com/apache/gobblin/pull/3855/files I made some optional fields mandatory. instrumentationEnabled was not one of them, but eventSubmitter was which is related.

instrumentationEnabled is usually true in prod, these variables are set false only for doing easier tests.

Also, in this particular case, instrumentationEnabled was hard coded to true in the above above constructor anyway.

umustafi · 2024-04-30T00:19:27Z

.../src/main/java/org/apache/gobblin/service/modules/utils/FlowCompilationValidationHelper.java

@@ -73,12 +74,12 @@ public class FlowCompilationValidationHelper {

  @Inject
  public FlowCompilationValidationHelper(Config config, SharedFlowMetricsSingleton sharedFlowMetricsSingleton,
-      UserQuotaManager userQuotaManager, FlowStatusGenerator flowStatusGenerator) {


how do we know the topologySpecFactory has been initialized by now?

TopologySpecFactory is guice based initialized

Yes I see that but what is the waiting you refer to in description "making spec compiler as a topology spec catalog listener and then wait for topology spec catalog to populate topologies." How do you know the topologySpecCatalog is populated?

Checking GobblinServiceManager start to see how TopologySpecCatalog is populated and it may not have the topologies loaded if it uses the listener method rather than TopologySpecFactory.getTopologies()

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

Line 508 in 33660f5

this.topologyCatalog.put(topologySpec);

One safe guard we can add is both Orchestrator and SpecCompiler'sonAddSpec method should check if specCompiler.isActive is true before accepting specs to pass to compiler.

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

Line 522 in 33660f5

this.orchestrator.getSpecCompiler().setActive(true);

SpecCompiler gets information about topology specs through orchestrator's onAddSpec. And orchestrator gets info about topology specs because it is a listener on topology spec catalog.
So compiler has to "wait for topology spec catalog to populate topologies". topology spec catalog is populated here https://github.com/apache/gobblin/blob/master/gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceManager.java#L508 but we may want compiler to have topology information before this code reaches.

umustafi · 2024-04-30T00:21:42Z

...service/src/main/java/org/apache/gobblin/service/modules/flow/BaseFlowToJobSpecCompiler.java

-
-  public BaseFlowToJobSpecCompiler(Config config, Optional<Logger> log, boolean instrumentationEnabled){
-    this.log = log.isPresent() ? log.get() : LoggerFactory.getLogger(getClass());
-    if (instrumentationEnabled) {


Arjun did some refactoring before which made instrumentation required I believe. Can you link that change to explain what's happening here?

phet

seems the intent is to remove the loose coupling between topo spec init and compiler init. are there any potential downsides? e.g. would it disallow the dynamic addition of additional topos during the course of system operation? this or any other limitations should be clearly captured in a comment

when you say:

topologies are static and fixed

is this essential to their nature or merely common practice we've adopted?

I'm not against a pragmatic change to preclude something we truly don't anticipate needing, but let's characterize for maintainers whether this is pure expedience (aka. easier than adding concurrency coordination to sub-service init)... or we actually believe it to be an essential part to modeling the solution (that the current impl mispercieved).

also nit: "waiting" in the title.

phet · 2024-05-01T18:39:49Z

...service/src/main/java/org/apache/gobblin/service/modules/flow/BaseFlowToJobSpecCompiler.java

    this.config = config;

-    /***
+    /*
     * ETL-5996


this is not an apache gobblin jira ticket

phet · 2024-05-01T18:40:40Z

...service/src/main/java/org/apache/gobblin/service/modules/flow/BaseFlowToJobSpecCompiler.java

-    } else if (addedSpec instanceof TopologySpec) {
-      return onAddTopologySpec( (TopologySpec) addedSpec);


is it wise to remove the ability to be a TS listener?

being a TS listener feels of no use to me. There should be no TS listener in gaas.

I equate topology specs w/ executors in the flow graph. since the FG can change dynamically w/o requiring a system restart, it doesn't seem out of the question to change the set of topos w/o a system restart either, given whatever new FG edges could indicate newly defined flow.edge.specExecutors

I agree that's not what we've done thus far, but it's arguably inconvenient to require two separate changes to define a new executor - one to the FG and one to the gaas configs. if it were possible to do both together, I would personally find that appealing

...src/test/java/org/apache/gobblin/service/modules/core/IdentityFlowToJobSpecCompilerTest.java

phet · 2024-05-01T18:46:43Z

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

@@ -114,6 +119,8 @@ public void setup() throws Exception {
    FlowLaunchHandler mockFlowTriggerHandler = mock(FlowLaunchHandler.class);
    DagManager mockDagManager = mock(DagManager.class);
    doNothing().when(mockDagManager).setTopologySpecMap(anyMap());
+    TopologySpecFactory mockedTopologySpecFactory = mock(TopologySpecFactory.class);
+    doReturn(Collections.singleton(this.topologySpec)).when(mockedTopologySpecFactory).getTopologies();


prefer the when().thenReturn() form, which is typesafe

I prefer this way, because in when(obj.method()).thenReturn(obj) it actually calls obj.method() which is rarely useful (and often generate exceptions) because obj is a mocked dummy object.

up to you. I do agree with a spy that doReturn is better for the reason you give, but with a mock, the type safety is helpful.

phet · 2024-05-01T18:48:43Z

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

-    // Make sure TopologyCatalog Listener is empty
-    Assert.assertTrue(specCompiler.getTopologySpecMap().size() == 0, "SpecCompiler should not know about any Topology "
-        + "before addition");
+    Assert.assertTrue(specCompiler.getTopologySpecMap().size() == 1, "SpecCompiler should know about any Topology "


desc here makes it sound like the test should be > 1. to that end, I didn't notice any Preconditions check or similar to insist on a non-empty collection. do we want one?

"SpecCompiler should know about any Topology irrespective of what is there in the topology catalog" ? how does that sound like test should be >1 ? What description do you suggest?

There is no precondition. SpecCompiler should just know about the topologies. Number of topologies known to spec compiler should just always be 1 (equals to total number of topologies).

"know about any" suggested it would apply even if there were > 1 topo specs. couldn't there be more than one?

also, if non-empty Set<TopologySpec> is critical, I was suggesting a guava Precondition in the ctor

phet · 2024-05-01T18:52:16Z

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

+    // Make sure TopologyCatalog empty
+    Assert.assertTrue(this.topologyCatalog.getSize() == 0, "Topology catalog should contain 0 Spec before addition");


seems this asserts that the spec compiler's topos may now potentially deviate from the topo catalog's, which might make the system harder to reason about... is it really a good thing to drop such an invariant?

there is a deviation only till everything is initialized. after that, they both should be same (1 in this test).

prior to this change, there's an "inconsistent timespan", where the compiler doesn't have topo specs initialized. you suggested this might compromise correctness of DagManagementStateStore.

after this change, sounds like there will be an "inconsistent timespan", where the TopologyCatalog may be out of sync w/ the topo specs known to the compiler. could there be any negative consequence to that?

arjun4084346 force-pushed the initTopology branch 3 times, most recently from 1dccd35 to 6d61f02 Compare April 29, 2024 12:42

arjun4084346 changed the title ~~initialize topology specs directly without waitging for listener call…~~ [GOBBLIN-2056] initialize topology specs directly without waitging for listener call… Apr 29, 2024

initialize topology specs directly without waitging for listener call…

5813b3f

…backs

arjun4084346 force-pushed the initTopology branch from 6d61f02 to 5813b3f Compare April 29, 2024 13:22

Will-Lo reviewed Apr 29, 2024

View reviewed changes

umustafi reviewed Apr 30, 2024

View reviewed changes

phet reviewed May 1, 2024

View reviewed changes

address review comments

03a3d6d

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[GOBBLIN-2056] initialize topology specs directly without waitging for listener call… #3937

[GOBBLIN-2056] initialize topology specs directly without waitging for listener call… #3937

arjun4084346 commented Apr 29, 2024 •

edited

Loading

Will-Lo left a comment

Will-Lo Apr 29, 2024

umustafi Apr 30, 2024

arjun4084346 May 1, 2024

umustafi Apr 30, 2024

arjun4084346 May 1, 2024

umustafi May 1, 2024

umustafi May 1, 2024

arjun4084346 May 1, 2024

umustafi Apr 30, 2024

phet left a comment •

edited

Loading

phet May 1, 2024

phet May 1, 2024

arjun4084346 May 1, 2024

phet May 1, 2024 •

edited

Loading

phet May 1, 2024

arjun4084346 May 1, 2024

phet May 1, 2024

phet May 1, 2024

arjun4084346 May 1, 2024

phet May 1, 2024 •

edited

Loading

phet May 1, 2024 •

edited

Loading

arjun4084346 May 1, 2024

phet May 1, 2024 •

edited

Loading

		} else if (addedSpec instanceof TopologySpec) {
		return onAddTopologySpec( (TopologySpec) addedSpec);

		// Make sure TopologyCatalog empty
		Assert.assertTrue(this.topologyCatalog.getSize() == 0, "Topology catalog should contain 0 Spec before addition");

[GOBBLIN-2056] initialize topology specs directly without waitging for listener call… #3937

Are you sure you want to change the base?

[GOBBLIN-2056] initialize topology specs directly without waitging for listener call… #3937

Conversation

arjun4084346 commented Apr 29, 2024 • edited Loading

JIRA

Description

Tests

Commits

Will-Lo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet May 1, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet May 1, 2024 • edited Loading

Choose a reason for hiding this comment

phet May 1, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phet May 1, 2024 • edited Loading

Choose a reason for hiding this comment

arjun4084346 commented Apr 29, 2024 •

edited

Loading

phet left a comment •

edited

Loading

phet May 1, 2024 •

edited

Loading

phet May 1, 2024 •

edited

Loading

phet May 1, 2024 •

edited

Loading

phet May 1, 2024 •

edited

Loading