[GOBBLIN-236] Add a ControlMessage injector as a RecordStreamProcessor #2090

htran1 · 2017-09-05T06:52:34Z

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-236

Description

Here are some details about my PR, including screenshots (if applicable):
Add a ControlMessage injector as a RecordStreamProcessor.
A ControlMessageInjector inspects an incoming record and can inject ControlMessages based on the content of the incoming record.

One use case for this is the injection of MetadataUpdateControlMessages when the latest schema has been updated to trigger the update of the schema used by other constructs downstream from the ControlMessageInjector.

Long running jobs are more likely to encounter issues due to the schema being updated after the start of the job.

Tests

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

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"

htran1 · 2017-09-05T16:14:22Z

@ibuenros, please review.

zxcware · 2017-09-07T22:55:59Z

Hi @htran1, could you add more description in the PR or the jira ticket about the motivation and general idea of control message injector? This can help reviewers obtain a direction of what to look for in terms of design and implementation.

ibuenros · 2017-09-06T17:38:02Z

gobblin-api/src/main/java/org/apache/gobblin/metadata/GlobalMetadata.java

+public class GlobalMetadata<S> implements Copyable<GlobalMetadata<S>> {
+  @Getter
+  private S schema;
+


It would be good to provide more functional methods:

/** Generate a copy of this object with a new schema. */ public GlobalMetadata<S> withSchema(S newSchema);

That way, when we add new attributes to GlobalMetadata, converters will require no change and still copy the correct metadata.

Okay, added a builder.

ibuenros · 2017-09-11T16:36:59Z

gobblin-api/src/main/java/org/apache/gobblin/converter/Converter.java

+                                      .getSchema(), workUnitState));
+                  out = new MetadataUpdateControlMessage<SO, DO>(this.outputGlobalMetadata);
+                }
+
                getMessageHandler().handleMessage((ControlMessage) in);


Should this receive in or out? If the decision is for in, then I would prefer this line be before the MetadataUpdateControlMessage block.

Okay, moved it before. I think in should be the one to be handled so that the handler gets a chance to handle the control message without modifications. The handler can choose to do schema conversion if it needs it.

ibuenros · 2017-09-11T16:38:58Z

gobblin-api/src/main/java/org/apache/gobblin/stream/ControlMessageInjector.java

+   * @param workUnitState a {@link WorkUnitState} object carrying configuration properties
+   * @return an initialized {@link ControlMessageInjector} instance
+   */
+  public ControlMessageInjector<SI, DI> init(WorkUnitState workUnitState) {


Can this be protected instead? Same with all methods except processStream.

ibuenros · 2017-09-11T16:42:43Z

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/Task.java

+    this.converter = closer.register(new MultiConverter(converters));
+
+    // can't have both record stream processors and converter lists configured
+    if (!this.recordStreamProcessors.isEmpty() && !converters.isEmpty()) {


Just to make it a bit clearer, can you do something like Preconditions.checkState(!this.recordStreamProcessor.isEmpty() ^ !this.converters.isEmpty(), "Mesasge...");

I need to call the closer here and both empty is valid so ^ won't work.

Added precondition check wrapped by a try catch block.

ibuenros · 2017-09-13T15:37:59Z

gobblin-api/src/main/java/org/apache/gobblin/stream/ControlMessageInjector.java

+                if (in instanceof MetadataUpdateControlMessage) {
+                  setInputGlobalMetadata(((MetadataUpdateControlMessage) in).getGlobalMetadata(),
+                      workUnitState);
+                  out = new MetadataUpdateControlMessage<SI, DI>(this.inputGlobalMetadata);


Why do you need a new output message? It seems to be identical to the input message.

setInputGlobalMetadata() can set this.inputGlobalMetadata to something other than the input metadata.

Removed inputGlobalMetadata from this class so that there is no confusion over the propagation of metadata. A ControlMessageInjector can store a modified copy of the metadata, but it won't be able to pass that change along to the next construct. Only converters should change schemas.

ibuenros · 2017-09-13T15:39:50Z

gobblin-api/src/main/java/org/apache/gobblin/converter/Converter.java

+
+                // update the output schema with the new input schema from the MetadataUpdateControlMessage
+                if (in instanceof MetadataUpdateControlMessage) {
+                  this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(),


Shouldn't this be GlobalMetadata.builderWithInput(in.getMetadata, Optional.of...)? (instead of inputStream)

Currently there are no other fields, so it either works. The original idea was to start with the inputStream GlobalMetadata and then overlay from in. But I think the suggestion of using in.getGlobalMetadata() is good since the copy will be handled in builderWithInput.

…lMessageInjector.

ibuenros · 2017-09-14T18:58:04Z

+1

Closes apache#2090 from htran1/control_message_injection

htran1 force-pushed the control_message_injection branch from 6957bd7 to 369bfe2 Compare September 6, 2017 18:36

[GOBBLIN-236] Add a ControlMessage injector as a RecordStreamProcessor

6206f8a

htran1 force-pushed the control_message_injection branch from 369bfe2 to 6206f8a Compare September 6, 2017 20:24

ibuenros suggested changes Sep 11, 2017

View reviewed changes

address review comments

2cb0e82

ibuenros suggested changes Sep 13, 2017

View reviewed changes

htran1 added 3 commits September 13, 2017 11:44

address review comments

0f264fa

Remove the inputGlobalMetadata from the base implementation of Contro…

e0c2fd4

…lMessageInjector.

remove unused imports

734de89

ibuenros approved these changes Sep 14, 2017

View reviewed changes

asfgit closed this in 3a03573 Sep 14, 2017

jack-moseley pushed a commit to jack-moseley/gobblin that referenced this pull request Sep 26, 2017

[GOBBLIN-236] Add a ControlMessage injector as a RecordStreamProcessor

fc89a7d

Closes apache#2090 from htran1/control_message_injection

zxliucmu pushed a commit to zxliucmu/incubator-gobblin that referenced this pull request Nov 16, 2017

[GOBBLIN-236] Add a ControlMessage injector as a RecordStreamProcessor

790b03d

Closes apache#2090 from htran1/control_message_injection

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[GOBBLIN-236] Add a ControlMessage injector as a RecordStreamProcessor #2090

[GOBBLIN-236] Add a ControlMessage injector as a RecordStreamProcessor #2090

htran1 commented Sep 5, 2017 •

edited

htran1 commented Sep 5, 2017

zxcware commented Sep 7, 2017

ibuenros Sep 6, 2017

htran1 Sep 11, 2017

ibuenros Sep 11, 2017

htran1 Sep 11, 2017

ibuenros Sep 11, 2017

htran1 Sep 11, 2017

ibuenros Sep 11, 2017

htran1 Sep 11, 2017

htran1 Sep 11, 2017

ibuenros Sep 13, 2017

htran1 Sep 13, 2017

htran1 Sep 13, 2017

ibuenros Sep 13, 2017

htran1 Sep 13, 2017

ibuenros commented Sep 14, 2017

[GOBBLIN-236] Add a ControlMessage injector as a RecordStreamProcessor #2090

[GOBBLIN-236] Add a ControlMessage injector as a RecordStreamProcessor #2090

Conversation

htran1 commented Sep 5, 2017 • edited

JIRA

Description

Tests

Commits

htran1 commented Sep 5, 2017

zxcware commented Sep 7, 2017

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

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

ibuenros commented Sep 14, 2017

htran1 commented Sep 5, 2017 •

edited