Only run normalization when needed #16794

alovew · 2022-09-15T23:54:31Z

Resolves #9672

Only run normalization if there are records that have been committed that have not yet been normalized

logging shouldRunNormalization method with tests update fix syncworkflowtests add activity version tests

colesnodgrass

Definitely need to change the Optional.get() == null check, I believe everything else is a recommendation.

colesnodgrass · 2022-09-21T01:20:02Z

airbyte-metrics/metrics-lib/src/main/java/io/airbyte/metrics/lib/MetricTags.java

+    if (origin != null) {
+      return origin.value();
+    } else {
+      return null;
+    }


I know this method is preexisting, but it feels unnecessary. The FailureOrigin is provided by the caller and this simply calls a method on the provided class?

Additionally this method is only called once:

for (final FailureReason reason : failureSummary.getFailures()) { MetricClientFactory.getMetricClient().count(OssMetricsRegistry.ATTEMPT_FAILED_BY_FAILURE_ORIGIN, 1, new MetricAttribute(MetricTags.FAILURE_ORIGIN, MetricTags.getFailureOrigin(reason.getFailureOrigin()))); }

Seems that we could move this functionality to this^ spot instead of having a dependency on MetricTags. Unless there is something I am missing here?

This was from when I was testing and running into odd errors - I'm not sure this actually ever happens and I can remove this since it's unrelated anyway!

colesnodgrass · 2022-09-21T01:22:25Z

...-persistence/src/main/java/io/airbyte/persistence/job/models/AttemptNormalizationStatus.java

+
+import java.util.Optional;
+
+public class AttemptNormalizationStatus {


Could this be replaced with a record class?
Should normalizationFailed be a boolean stead, forcing a true/false to be provided?

colesnodgrass · 2022-09-21T01:26:03Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+  @Inject
+  private JobPersistence jobPersistence;


Per both my preference and Micronaut's preference

Field injection makes it harder to understand a class’s requirements, making it easier to get a NullPointerException when testing a class using Field Injection. We recommend you use Constructor Injection.

could we change this to be constructor injected?

Also, the @AllArgsConstructor will create a constructor that wins over the field injection. This is probably working because we don't have any qualifiers or configuration properties here. However, per @colesnodgrass's request, we should use explicit constructors here and avoid Lombok/field injection to make the code easier to read.

colesnodgrass · 2022-09-21T01:26:38Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+
+  @Override
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
+  public Boolean shouldRunNormalization(final Long jobId, final Long attemptNumber, final Optional<Long> numCommittedRecords) throws IOException {


Should this return boolean instead? It appears we only return true/false and never null.

colesnodgrass · 2022-09-21T01:32:52Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+          // if normalization failed on past attempt,
+          // add number of records committed on that attempt to
+          // total committed number
+          if (n.getNormalizationFailed()) {


If getNormalizationFailed returned a boolean instead of a Boolean, this if statement could be removed, as the above if statement already verified it was false.

colesnodgrass · 2022-09-21T01:37:00Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+    }
+
+    final List<AttemptNormalizationStatus> attemptNormalizationStatuses = jobPersistence.getAttemptNormalizationStatusesForJob(jobId);
+    final AtomicReference<Long> totalRecordsCommitted = new AtomicReference<>(0L);


could be an AtomicLong

colesnodgrass · 2022-09-21T01:37:26Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+
+    final List<AttemptNormalizationStatus> attemptNormalizationStatuses = jobPersistence.getAttemptNormalizationStatusesForJob(jobId);
+    final AtomicReference<Long> totalRecordsCommitted = new AtomicReference<>(0L);
+    final AtomicReference<Boolean> shouldReturnTrue = new AtomicReference<>(false);


could be an AtomicBoolean

colesnodgrass · 2022-09-21T01:40:54Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+            if (n.getRecordsCommitted().get() == null) {
+              shouldReturnTrue.set(true);
+              return;
+            } else if (n.getRecordsCommitted().get() != 0L) {
+              totalRecordsCommitted.set(totalRecordsCommitted.get() + n.getRecordsCommitted().get());
+            }


Calling Optional.get() == null doesn't work as Optional.get() throws an NPE if the optional is empty.

Maybe replace this with Optional.ifPresentOrElse?

n.getRecordsComitted.ifPresentOrElse( // can use addAndGet if using an AtomicLong numRecords -> totalRecordsComitted.addAndGet(numRecords), () -> shouldReturnTrue.set(true) );

colesnodgrass · 2022-09-21T20:51:42Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+          // were committed records and run normalization
+          if (n.recordsCommitted().isEmpty()) {
+            shouldReturnTrue.set(true);
+            return;


this return is not needed here as pointed out by my IDE

return is unnecessary as the last statement in a void method

the return is to beak out of the iteration block instead of returning a value from the method overall

colesnodgrass · 2022-09-21T20:57:19Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+            shouldReturnTrue.set(true);
+            return;
+          } else if (n.recordsCommitted().get() != 0L) {
+            totalRecordsCommitted.set(totalRecordsCommitted.get() + n.recordsCommitted().get());


.set should be replaced with .addAndGet(n.recordsCommitted().get()) as this will avoids the totalRecordsCommitted.get() call

colesnodgrass

one recommended change from the set call to addAndGet, otherwise 👍🏻

gosusnp · 2022-09-21T22:56:24Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+
+  private final Optional<JobPersistence> jobPersistence;
+
+  public NormalizationSummaryCheckActivityImpl(final Optional<JobPersistence> jobPersistence) {


What's the use case for the optional JobPersistence?
If it's related to the data plane worker, feels like we may want to have a follow up to be able to use API instead of direct DB access here, otherwise multi-cloud will run into this issue again.

yes that's exactly the reason - i agree, didn't realize while I was writing this but changing this to use the API instead later makes sense. It's not hugely crucial in this case since all that will happen is that normalization will run when it doesn't need to

In the end, this PR is still a step forward. I think this also highlights the fact that we need to focus on removing direct DB access from the workers.

cc @jdpgrailsdev in case you were keeping track of the many reasons we should be doing this ;)

gosusnp

Looks good, just want to double check on how we skip normalization, it might be a bit aggressive but I am not that familiar with the following steps.

gosusnp · 2022-09-21T23:50:59Z

...rs/src/main/java/io/airbyte/workers/temporal/sync/NormalizationSummaryCheckActivityImpl.java

+
+    // if the count of committed records for this attempt is > 0 OR if it is null,
+    // then we should run normalization
+    if (numCommittedRecords.get() == null || numCommittedRecords.get() > 0) {


nit: the null check should be numCommittedRecords.isEmpty() or .isPresent()

gosusnp · 2022-09-21T23:57:06Z

airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/SyncWorkflowImpl.java

+            }
+            if (!shouldRun) {
+              LOGGER.info("Skipping normalization because there are no records to normalize.");
+              break;


Are we sure we want to break in this case? I think there may be other steps after normalization, like dbt for example.

Only run normalization when records have been committed

github-actions bot added area/platform issues related to the platform area/worker Related to worker labels Sep 15, 2022

alovew temporarily deployed to more-secrets September 15, 2022 23:56 Inactive

github-actions bot added the area/scheduler label Sep 19, 2022

alovew temporarily deployed to more-secrets September 19, 2022 17:59 Inactive

alovew temporarily deployed to more-secrets September 19, 2022 21:56 Inactive

alovew force-pushed the anne/check-before-running-normalization branch from 9a6df23 to c415820 Compare September 20, 2022 17:51

alovew temporarily deployed to more-secrets September 20, 2022 17:53 Inactive

alovew temporarily deployed to more-secrets September 20, 2022 21:34 Inactive

alovew marked this pull request as ready for review September 20, 2022 22:48

alovew temporarily deployed to more-secrets September 20, 2022 22:50 Inactive

alovew added 2 commits September 20, 2022 15:56

add logic

2420956

logging shouldRunNormalization method with tests update fix syncworkflowtests add activity version tests

changes after rebase

3976351

alovew force-pushed the anne/check-before-running-normalization branch from ddedce3 to 3976351 Compare September 20, 2022 23:10

github-actions bot removed the area/scheduler label Sep 20, 2022

alovew temporarily deployed to more-secrets September 20, 2022 23:12 Inactive

remove logs and more comments

7b92854

alovew requested review from jdpgrailsdev, benmoriceau, gosusnp and colesnodgrass September 20, 2022 23:14

alovew temporarily deployed to more-secrets September 20, 2022 23:16 Inactive

alovew temporarily deployed to more-secrets September 21, 2022 01:03 Inactive

colesnodgrass requested changes Sep 21, 2022

View reviewed changes

code review comments

84223f1

alovew requested a review from colesnodgrass September 21, 2022 19:31

alovew temporarily deployed to more-secrets September 21, 2022 19:32 Inactive

remove get methods

e893757

alovew temporarily deployed to more-secrets September 21, 2022 20:36 Inactive

colesnodgrass reviewed Sep 21, 2022

View reviewed changes

colesnodgrass approved these changes Sep 21, 2022

View reviewed changes

use addandget

64509b8

alovew temporarily deployed to more-secrets September 21, 2022 21:21 Inactive

gosusnp reviewed Sep 21, 2022

View reviewed changes

alovew merged commit f28d5f3 into master Sep 21, 2022

alovew deleted the anne/check-before-running-normalization branch September 21, 2022 23:50

gosusnp approved these changes Sep 21, 2022

View reviewed changes

This was referenced Sep 22, 2022

Bump Airbyte version from 0.40.8 to 0.40.9 #17060

Closed

Bump Airbyte version from 0.40.8 to 0.40.9 #17093

Merged

robbinhan pushed a commit to robbinhan/airbyte that referenced this pull request Sep 29, 2022

Only run normalization when needed (airbytehq#16794)

2a589e7

Only run normalization when records have been committed

jhammarstedt pushed a commit to jhammarstedt/airbyte that referenced this pull request Oct 31, 2022

Only run normalization when needed (airbytehq#16794)

3116421

Only run normalization when records have been committed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Only run normalization when needed #16794

Only run normalization when needed #16794

alovew commented Sep 15, 2022 •

edited

Loading

colesnodgrass left a comment

colesnodgrass Sep 21, 2022

alovew Sep 21, 2022

colesnodgrass Sep 21, 2022

colesnodgrass Sep 21, 2022

jdpgrailsdev Sep 21, 2022

colesnodgrass Sep 21, 2022

colesnodgrass Sep 21, 2022

colesnodgrass Sep 21, 2022

colesnodgrass Sep 21, 2022

colesnodgrass Sep 21, 2022

colesnodgrass Sep 21, 2022

alovew Sep 21, 2022

colesnodgrass Sep 21, 2022

colesnodgrass left a comment

gosusnp Sep 21, 2022

alovew Sep 21, 2022

gosusnp Sep 21, 2022

gosusnp left a comment

gosusnp Sep 21, 2022

gosusnp Sep 21, 2022


		import java.util.Optional;

		public class AttemptNormalizationStatus {


		private final Optional<JobPersistence> jobPersistence;

		public NormalizationSummaryCheckActivityImpl(final Optional<JobPersistence> jobPersistence) {

Only run normalization when needed #16794

Only run normalization when needed #16794

Conversation

alovew commented Sep 15, 2022 • edited Loading

colesnodgrass 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

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

colesnodgrass 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

gosusnp left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alovew commented Sep 15, 2022 •

edited

Loading