[GOBBLIN-423] Add record count limit to salesforce source #2300

yukuai518 · 2018-03-08T23:51:36Z

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

Description

Here are some details about my PR, including screenshots (if applicable):
Set fixed upper bound in bootstrapping mode. Don't move the upper bound in the future run.

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"

yukuai518 · 2018-03-09T00:41:27Z

@zxcware @htran1 Please review

htran1 · 2018-03-08T23:56:34Z

gobblin-core/src/main/java/org/apache/gobblin/source/extractor/partition/Partitioner.java

      // The upper bounds can be removed for last work unit
      partitions.add(new Partition(lowWatermark, highWatermark, true, false));
    } else {
-      // The upper bounds can not be removed for last work unit
+      // The upper bounds can not be removed for last work unit (even for bootstrapping mode, we want to fix the upperbound so next time we would know if we have catched up)


catched -> caught

htran1 · 2018-03-09T04:06:04Z

gobblin-core/src/main/java/org/apache/gobblin/source/extractor/partition/Partitioner.java

-    } else {
-      // The upper bounds can not be removed for last work unit
+
+    if (isBootStrapWithFixedHighWatermark()) {


This can be merged with the last else case. So if (!isBootStrapWithFixedHighWatermark() && (isFullDump() || isSnapshot(extractType)))

htran1 · 2018-03-09T07:21:55Z

gobblin-core/src/main/java/org/apache/gobblin/source/extractor/partition/Partitioner.java

@@ -322,6 +333,17 @@ protected long getLowWatermark(ExtractType extractType, WatermarkType watermarkT
        lowWatermark = this.getAppendLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark);
      }
    }
+
+    if (isBootStrapWithFixedHighWatermark()) {


This looks the same as the else block above, so it can be merged by changing the if to if (!isBootStrapWithFixedHighWatermark() && (this.isFullDump() || this.isWatermarkOverride())) {

htran1 · 2018-03-09T07:31:28Z

gobblin-core/src/main/java/org/apache/gobblin/source/extractor/partition/Partitioner.java

+        highWatermark = this.state.getPropAsLong(ConfigurationKeys.SOURCE_QUERYBASED_END_VALUE);
+        LOG.info("Bootstrapping mode: persisted high watermark " + highWatermark);
+      } else {
+        this.state.setProp(ConfigurationKeys.SOURCE_QUERYBASED_END_VALUE, highWatermark);


Is this state getting persisted and propagated to the next run? If it does then what happens when user changes the configuration? Will the config change override the state?

htran1 · 2018-03-09T07:33:20Z

gobblin-salesforce/src/main/java/org/apache/gobblin/salesforce/SalesforceSource.java

@@ -79,6 +79,11 @@
  public static final boolean DEFAULT_USE_ALL_OBJECTS = false;

  private static final String ENABLE_DYNAMIC_PARTITIONING = "salesforce.enableDynamicPartitioning";
+  private static final String DYNAMIC_PROBING_TOTAL_RECORDS_LIMIT = "salesforce.dynamicProbing.total.records.limit";


Camel case names should be used here.

htran1 · 2018-03-09T07:43:55Z

gobblin-salesforce/src/main/java/org/apache/gobblin/salesforce/SalesforceSource.java

+        break;
+      }
+      histogramActual.add(group);
+      if (histogramActual.getGroups().size() >= state.getPropAsLong(DYNAMIC_PROBING_TOTAL_BUCKETS_LIMIT, DYNAMIC_PROBING_DEFAULT_TOTAL_BUCKETS_LIMIT)) {


Why have the bucket limit here?

@htran1 do we have total query limits? I think each bucket is a single query, if we have total limits for querying count, we can set this bucket number.

The query limit is enforced by salesforce.dynamicProbingLimit. The loop here doesn't execute any queries, so the limit is not required here.

htran1 · 2018-03-09T07:45:36Z

gobblin-salesforce/src/main/java/org/apache/gobblin/salesforce/SalesforceSource.java

@@ -131,7 +136,19 @@
    Partition partition = new Partitioner(state).getGlobalPartition(previousWatermark);
    Histogram histogram = getHistogram(sourceEntity.getSourceEntityName(), watermarkColumn, state, partition);

-    String specifiedPartitions = generateSpecifiedPartitions(histogram, minTargetPartitionSize, maxPartitions,
+    // we should look if the count is too big, cut off early if count exceeds the limit, or bucket size is too large
+    Histogram histogramActual = new Histogram();


We should consider pushing this logic into getHistogram() for early termination of the search or add a note here indicating possible future work.

htran1 · 2018-03-13T16:38:41Z

gobblin-salesforce/src/main/java/org/apache/gobblin/salesforce/SalesforceSource.java

+        break;
+      }
+      histogramActual.add(group);
+      if (histogramActual.getGroups().size() >= state.getPropAsLong(DYNAMIC_PROBING_TOTAL_BUCKETS_LIMIT, DYNAMIC_PROBING_DEFAULT_TOTAL_BUCKETS_LIMIT)) {


The query limit is enforced by salesforce.dynamicProbingLimit. The loop here doesn't execute any queries, so the limit is not required here.

htran1 · 2018-03-13T16:41:44Z

gobblin-salesforce/src/main/java/org/apache/gobblin/salesforce/SalesforceSource.java

@@ -504,6 +534,14 @@ void add(Histogram histogram) {
      totalRecordCount += histogram.totalRecordCount;
    }

+    HistogramGroup get(int idx) {


Why is this required? I don't see anywhere where the null result is checked.

I use this to get the last group.

htran1 · 2018-03-21T22:15:03Z

gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobScheduler.java

+    public Void call() throws JobException {
+      try {
+        while (true) {
+          currentJobLauncher = buildJobLauncher(jobProps);;


Extra semi-colon.

htran1 · 2018-03-21T22:17:16Z

gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobScheduler.java

@@ -281,18 +271,104 @@ protected void startServices() throws Exception {
  @Override
  public void runJob(Properties jobProps, JobListener jobListener) throws JobException {
    try {
-      JobLauncher jobLauncher = buildGobblinHelixJobLauncher(jobProps);
-      runJob(jobProps, jobListener, jobLauncher);
+      while (true) {


Create and call the RetriggeringJobCallable call here.

htran1 · 2018-03-21T22:21:31Z

gobblin-core-base/src/main/java/org/apache/gobblin/source/extractor/extract/AbstractSource.java

@@ -139,4 +139,8 @@
  public Extract createExtract(TableType type, String namespace, String table) {
    return this.extractFactory.getUniqueExtract(type, namespace, table);
  }
+
+  public boolean isRetriggerRequired() {


Add @OverRide.

htran1 · 2018-03-21T22:42:41Z

gobblin-salesforce/src/main/java/org/apache/gobblin/salesforce/SalesforceSource.java

+      histogramAdjust = new Histogram();
+      for (HistogramGroup group : histogram.getGroups()) {
+        if (histogramAdjust.getTotalRecordCount() + group.count > state
+            .getPropAsLong(DYNAMIC_PROBING_TOTAL_RECORDS_LIMIT, DYNAMIC_PROBING_DEFAULT_TOTAL_RECORDS_LIMIT)) {


Need a different name here related to early stop.

htran1 · 2018-03-21T23:48:59Z

gobblin-core/src/main/java/org/apache/gobblin/source/extractor/partition/Partitioner.java

    List<Partition> partitions = new ArrayList<>();

    List<String> watermarkPoints = state.getPropAsList(USER_SPECIFIED_PARTITIONS);
+    boolean isEarlyStop = state.getPropAsBoolean(IS_EARLY_STOP);
+
+    if (isEarlyStop && isEarlyStopEnabled() && isFullDump()) {


Always raise the configuration error and do it earlier.

htran1 · 2018-03-21T23:49:50Z

gobblin-core/src/main/java/org/apache/gobblin/source/extractor/partition/Partitioner.java

-    if (isFullDump() || isSnapshot(extractType)) {
+
+    // If it is early stop, we should not remove upper bounds
+    if ((isFullDump() || isSnapshot(extractType)) && !(isEarlyStop && isEarlyStopEnabled())) {


Only check isEarlyStop since it will always be false when isEarlyStopEnabled() is false.

htran1 · 2018-03-21T23:50:27Z

gobblin-core/src/main/java/org/apache/gobblin/source/extractor/partition/Partitioner.java

   */
  @VisibleForTesting
  protected long getLowWatermark(ExtractType extractType, WatermarkType watermarkType, long previousWatermark,
      int deltaForNextWatermark) {
    long lowWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE;
-    if (this.isFullDump() || this.isWatermarkOverride()) {
+
+    if ((this.isFullDump() || this.isWatermarkOverride()) && !this.isEarlyStopEnabled()) {


Remove isEarlyStopEnabled() and disallow this config for full dump and watermark override.

htran1 · 2018-03-22T00:22:19Z

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

@@ -887,6 +887,10 @@ private void cleanupStagingData(JobState jobState)
    }
  }

+  public boolean isRetriggerRequired() {


I think we should expose only the early stop state and have the retrigger decision made in the scheduler.

htran1

+1

Closes apache#2300 from yukuai518/limit

htran1 suggested changes Mar 9, 2018

View reviewed changes

yukuai518 force-pushed the limit branch from e420a33 to bdd8a74 Compare March 10, 2018 00:41

htran1 suggested changes Mar 13, 2018

View reviewed changes

yukuai518 changed the title ~~Add record count limit to salesforce source~~ [GOBBLIN-423] Add record count limit to salesforce source Mar 21, 2018

yukuai518 force-pushed the limit branch from bdd8a74 to ce3c8d9 Compare March 21, 2018 17:08

htran1 suggested changes Mar 21, 2018

View reviewed changes

htran1 suggested changes Mar 22, 2018

View reviewed changes

yukuai518 added 9 commits March 22, 2018 14:57

Add early stop and retriggering feature for Salesforce source

e162342

Propogate early stop flag from SourceDecor

3335b1e

Resolve comments

d7e54bd

Resolve comments

74b4313

Add override alias

badedcf

Fix build issue

47531a1

Fix documentation

232c5af

Add low watermark log

be8a26c

Resolve conflict

e10927e

yukuai518 force-pushed the limit branch from f7a9235 to e10927e Compare March 22, 2018 22:39

htran1 approved these changes Mar 22, 2018

View reviewed changes

asfgit closed this in 5b8f7df Mar 22, 2018

tilakpatidar pushed a commit to tilakpatidar/incubator-gobblin that referenced this pull request Aug 10, 2018

[GOBBLIN-423] Add record count limit to salesforce source

53cae80

Closes apache#2300 from yukuai518/limit

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[GOBBLIN-423] Add record count limit to salesforce source #2300

[GOBBLIN-423] Add record count limit to salesforce source #2300

yukuai518 commented Mar 8, 2018

yukuai518 commented Mar 9, 2018

htran1 Mar 8, 2018

htran1 Mar 9, 2018

htran1 Mar 9, 2018

htran1 Mar 9, 2018

htran1 Mar 9, 2018

htran1 Mar 9, 2018

yukuai518 Mar 9, 2018

htran1 Mar 13, 2018

htran1 Mar 9, 2018

htran1 Mar 13, 2018

htran1 Mar 13, 2018

yukuai518 Mar 21, 2018 •

edited

htran1 Mar 21, 2018

htran1 Mar 21, 2018

htran1 Mar 21, 2018

htran1 Mar 21, 2018

htran1 Mar 21, 2018

htran1 Mar 21, 2018

htran1 Mar 21, 2018

htran1 Mar 22, 2018

htran1 left a comment

[GOBBLIN-423] Add record count limit to salesforce source #2300

[GOBBLIN-423] Add record count limit to salesforce source #2300

Conversation

yukuai518 commented Mar 8, 2018

JIRA

Description

Tests

Commits

yukuai518 commented Mar 9, 2018

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

yukuai518 Mar 21, 2018 • edited

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

htran1 left a comment

Choose a reason for hiding this comment

yukuai518 Mar 21, 2018 •

edited