refresh before syncs when feature flag is on #19888

alovew · 2022-11-29T19:20:19Z

Refresh schemas before each sync using feature flag

The main logic here is in SyncWorkflowImpl.java

jdpgrailsdev · 2022-11-30T21:07:18Z

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

+    try {
+      airbyteApiClient.getSourceApi().discoverSchemaForSource(requestBody);
+    } catch (final Exception e) {
+      log.info("Attempted schema refresh, but failed.");


This should probably be an error log with the exception itself logged as well.

oops yep, meant to update that! I also have a question about logging in general. These logs (and the ones in SyncWorkflowImpl) are not showing up in the UI logs and I'm not sure why.

alovew · 2022-12-01T20:17:17Z

Two questions on this. I'm seeing this CI error:

___________ test_read_stream_returns_error_if_stream_does_not_exist ____________

    def test_read_stream_returns_error_if_stream_does_not_exist():
        expected_status_code = 400
        expected_detail = "Could not perform read with with error: \"The requested stream not_in_manifest was not found in the source. Available streams: dict_keys(['hashiras', 'breathing-techniques'])\""
    
        api = DefaultApiImpl()
        loop = asyncio.get_event_loop()
        with pytest.raises(HTTPException) as actual_exception:
            loop.run_until_complete(
                api.read_stream(StreamReadRequestBody(manifest=MANIFEST, config=***, stream="not_in_manifest"))
            )
    
        assert actual_exception.value.status_code == expected_status_code
>       assert expected_detail in actual_exception.value.detail
E       assert 'Could not perform read with with error: "The requested stream not_in_manifest was not found in the source. Available streams: dict_keys([\'hashiras\', \'breathing-techniques\'])"' in "Could not perform read with with error: The requested stream not_in_manifest was not found in the source. Available streams: dict_keys(['hashiras', 'breathing-techniques'])"
E        +  where "Could not perform read with with error: The requested stream not_in_manifest was not found in the source. Available streams: dict_keys(['hashiras', 'breathing-techniques'])" = HTTPException(status_code=400, detail="Could not perform read with with error: The requested stream not_in_manifest was not found in the source. Available streams: dict_keys(['hashiras', 'breathing-techniques'])").detail
E        +    where HTTPException(status_code=400, detail="Could not perform read with with error: The requested stream not_in_manifest was not found in the source. Available streams: dict_keys(['hashiras', 'breathing-techniques'])") = <ExceptionInfo HTTPException(status_code=400, detail="Could not perform read with with error: The requested stream not_in_manifest was not found in the source. Available streams: dict_keys(['hashiras', 'breathing-techniques'])") tblen=3>.value

and I'm not seeing how that's related.

Also - the logs I added in SyncWorkflowImpl are not showing up in the UI. Does anyone have insight into which logs end up in the UI?

benmoriceau · 2022-12-01T20:51:11Z

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

@@ -72,6 +86,23 @@ public StandardSyncOutput run(final JobRunConfig jobRunConfig,

    final int version = Workflow.getVersion(VERSION_LABEL, Workflow.DEFAULT_VERSION, CURRENT_VERSION);
    final String taskQueue = Workflow.getInfo().getTaskQueue();
+
+    if (version > Workflow.DEFAULT_VERSION) {


We need a dedicated version for that. Re-using the existing version won't work.

oh I actually had this locally but didn't push up the change - doing that now

benmoriceau · 2022-12-01T20:52:31Z

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


  @Trace(operationName = WORKFLOW_TRACE_OPERATION_NAME)
  @Override
  public StandardSyncOutput run(final JobRunConfig jobRunConfig,
                                final IntegrationLauncherConfig sourceLauncherConfig,
                                final IntegrationLauncherConfig destinationLauncherConfig,
                                final StandardSyncInput syncInput,
-                                final UUID connectionId) {
+                                final UUID connectionId)
+      throws JsonValidationException, ConfigNotFoundException, IOException, ApiException {


I think that those exception shouldn't be in the method signature, we should catch them and set the failure reason in the return StandardSyncOutput

Just updated this -- I actually think we shouldn't fail the sync just because a schema refresh fails. I logged the error, but I'm not sure if that would be confusing to users. What do you think?

gosusnp · 2022-12-02T22:57:31Z

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


-  public RefreshSchemaActivityImpl(Optional<ConfigRepository> configRepository, SourceApi sourceApi) {
+  public RefreshSchemaActivityImpl(Optional<ConfigRepository> configRepository,
+                                   AirbyteApiClient airbyteApiClient,


Can we inject the specific ApiClient that are needed rather than the top level one?

I tried this but was getting a micronaut bean error, so I followed how you did it in the PersistStateActivity

At the time PersistStateActivity was added, I don't think we had the independent clients yet.
The way to solve the micronaut bean error should be to add the client to the API factory:

airbyte/airbyte-workers/src/main/java/io/airbyte/workers/config/ApiClientBeanFactory.java

Line 38 in 8c780bd

public AirbyteApiClient airbyteApiClient(

@benmoriceau, any specific reasons we do not have those yet other than did not have time to get to it?

Not that I know of. We should add them

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

gosusnp · 2022-12-02T23:04:43Z

airbyte-workers/src/test/java/io/airbyte/workers/temporal/sync/SyncWorkflowTest.java

@@ -423,4 +450,13 @@ private static void verifyDbtTransform(final DbtTransformationActivity dbtTransf
        operatorDbtInput);
  }

+  private static void verifyShouldRefreshSchema(final RefreshSchemaActivity refreshSchemaActivity) throws IOException {


airbyte-workers/src/main/java/io/airbyte/workers/config/ApiClientBeanFactory.java

gosusnp · 2022-12-06T00:27:58Z

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

+    try {
+      sourceApi.discoverSchemaForSource(requestBody);
+    } catch (final Exception e) {
+      log.error("Attempted schema refresh, but failed with error: ", e);


nit: Could be worth adding a comment explaining that we swallow the exception here to avoid blocking the replication if we fail to refresh.

implemented

This reverts commit 2b045a9.

octavia-squidington-iv added area/platform issues related to the platform area/worker Related to worker labels Nov 29, 2022

alovew temporarily deployed to more-secrets November 29, 2022 20:01 Inactive

alovew temporarily deployed to more-secrets November 29, 2022 22:46 Inactive

alovew force-pushed the anne/feature-flag-refresh-schema branch from 0a83a93 to 25cc417 Compare November 30, 2022 18:19

alovew temporarily deployed to more-secrets November 30, 2022 18:22 Inactive

octavia-squidington-iv added the area/server label Nov 30, 2022

alovew temporarily deployed to more-secrets November 30, 2022 18:59 Inactive

alovew force-pushed the anne/feature-flag-refresh-schema branch from 9326cc7 to 527d2fc Compare November 30, 2022 20:09

alovew requested review from benmoriceau, jdpgrailsdev, gosusnp and colesnodgrass November 30, 2022 20:10

alovew temporarily deployed to more-secrets November 30, 2022 20:12 Inactive

jdpgrailsdev reviewed Nov 30, 2022

View reviewed changes

alovew temporarily deployed to more-secrets November 30, 2022 21:55 Inactive

octavia-squidington-iv removed the area/server label Nov 30, 2022

alovew temporarily deployed to more-secrets November 30, 2022 22:02 Inactive

alovew temporarily deployed to more-secrets November 30, 2022 22:05 Inactive

alovew temporarily deployed to more-secrets November 30, 2022 22:07 Inactive

alovew requested a review from jdpgrailsdev December 1, 2022 19:41

benmoriceau previously requested changes Dec 1, 2022

View reviewed changes

alovew force-pushed the anne/feature-flag-refresh-schema branch from a69ad8c to 7db8ba9 Compare December 1, 2022 21:00

alovew temporarily deployed to more-secrets December 1, 2022 21:03 Inactive

alovew requested a review from benmoriceau December 1, 2022 21:17

alovew added 5 commits December 2, 2022 14:24

Catch errors

6d99d6c

tests

9b013fe

fix test

f09f87b

update tests

3fc3751

fix test

825e42a

alovew force-pushed the anne/feature-flag-refresh-schema branch from da9aeb6 to 825e42a Compare December 2, 2022 22:24

alovew temporarily deployed to more-secrets December 2, 2022 22:27 Inactive

gosusnp reviewed Dec 2, 2022

View reviewed changes

alovew requested a review from gosusnp December 2, 2022 23:31

use sourceApi directly

a35fc50

alovew temporarily deployed to more-secrets December 3, 2022 00:31 Inactive

fix test

6362a97

alovew temporarily deployed to more-secrets December 3, 2022 00:56 Inactive

formatting

ed7c615

alovew temporarily deployed to more-secrets December 3, 2022 01:19 Inactive

gosusnp reviewed Dec 5, 2022

View reviewed changes

airbyte-workers/src/main/java/io/airbyte/workers/config/ApiClientBeanFactory.java Outdated Show resolved Hide resolved

Move error handling into activities

c3faa67

alovew temporarily deployed to more-secrets December 5, 2022 23:08 Inactive

alovew requested a review from gosusnp December 5, 2022 23:13

separate singleton for apiclient

6b70630

alovew temporarily deployed to more-secrets December 5, 2022 23:51 Inactive

gosusnp approved these changes Dec 6, 2022

View reviewed changes

add comments

c98cd7b

alovew temporarily deployed to more-secrets December 6, 2022 01:02 Inactive

alovew merged commit 2b045a9 into master Dec 6, 2022

alovew deleted the anne/feature-flag-refresh-schema branch December 6, 2022 02:46

This was referenced Dec 6, 2022

Bump Airbyte version from 0.40.23 to 0.40.24 #20161

Closed

Bump Airbyte version from 0.40.23 to 0.40.24 #20206

Merged

malikdiarra added a commit that referenced this pull request Dec 9, 2022

Revert "refresh before syncs when feature flag is on (#19888)"

02f7e86

This reverts commit 2b045a9.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refresh before syncs when feature flag is on #19888

refresh before syncs when feature flag is on #19888

alovew commented Nov 29, 2022 •

edited

Loading

jdpgrailsdev Nov 30, 2022

alovew Nov 30, 2022

alovew commented Dec 1, 2022

benmoriceau Dec 1, 2022

alovew Dec 1, 2022

benmoriceau Dec 1, 2022

alovew Dec 1, 2022

gosusnp Dec 2, 2022

alovew Dec 2, 2022

gosusnp Dec 2, 2022

benmoriceau Dec 3, 2022 •

edited

Loading

gosusnp Dec 2, 2022

gosusnp Dec 6, 2022

refresh before syncs when feature flag is on #19888

refresh before syncs when feature flag is on #19888

Conversation

alovew commented Nov 29, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alovew commented Dec 1, 2022

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

benmoriceau Dec 3, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alovew commented Nov 29, 2022 •

edited

Loading

benmoriceau Dec 3, 2022 •

edited

Loading