🐛 Do not propagate exceptions on source close. #33609

davinchia · 2023-12-18T22:54:56Z

What

Based on #33211 , we see certain networking set ups can throw exceptions as we close up. This gives the false perception the sync is failed even though we've moved all data.

Further, we cannot actually action on these exceptions since:

we are going to close down anyway.
there is minimal impact all round since we run on containerised infrastructure and an exception closing the db connection likely means the connection was already closed on the DB side (at least the cases we've observed in practice).

How

Describe the solution

🚨 User Impact 🚨

Are there any breaking changes? What is the end result perceived by the user?

For connector PRs, use this section to explain which type of semantic versioning bump occurs as a result of the changes. Refer to our Semantic Versioning for Connectors guidelines for more information. Breaking changes to connectors must be documented by an Airbyte engineer (PR author, or reviewer for community PRs) by using the Breaking Change Release Playbook.

If there are breaking changes, please merge this PR with the 🚨🚨 emoji so changelog authors can further highlight this if needed.

Pre-merge Actions

Expand the relevant checklist and delete the others.

New Connector

Community member or Airbyter

Community member? Grant edit access to maintainers (instructions)
Unit & integration tests added and passing. Community members, please provide proof of success locally e.g: screenshot or copy-paste unit, integration, and acceptance test output. To run acceptance tests for a Python connector, follow instructions in the README. For java connectors run ./gradlew :airbyte-integrations:connectors:<name>:integrationTest.
Connector version is set to 0.0.1
- Dockerfile has version 0.0.1
Documentation updated
- Connector's README.md
- Connector's bootstrap.md. See description and examples
- docs/integrations/<source or destination>/<name>.md including changelog with an entry for the initial version. See changelog example
- docs/integrations/README.md

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

Create a non-forked branch based on this PR and test the below items on it
Build is successful
If new credentials are required for use in CI, add them to GSM. Instructions.

Updating a connector

Community member or Airbyter

Grant edit access to maintainers (instructions)
Unit & integration tests added

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

Create a non-forked branch based on this PR and test the below items on it
Build is successful
If new credentials are required for use in CI, add them to GSM. Instructions.

Connector Generator

Issue acceptance criteria met
PR name follows PR naming conventions
If adding a new generator, add it to the list of scaffold modules being tested
The generator test modules (all connectors with -scaffold in their name) have been updated with the latest scaffold by running ./gradlew :airbyte-integrations:connector-templates:generator:generateScaffolds then checking in your changes
Documentation which references the generator is updated as needed

Updating the Python CDK

Airbyter

Before merging:

Pull Request description explains what problem it is solving
Code change is unit tested
Build and my-py check pass
Smoke test the change on at least one affected connector
- On Github: Run this workflow, passing --use-local-cdk --name=source-<connector> as options
- Locally: airbyte-ci connectors --use-local-cdk --name=source-<connector> test
PR is reviewed and approved

After merging:

Publish the CDK
- The CDK does not follow proper semantic versioning. Choose minor if this the change has significant user impact or is a breaking change. Choose patch otherwise.
- Write a thoughtful changelog message so we know what was updated.
Merge the platform PR that was auto-created for updating the Connector Builder's CDK version
- This step is optional if the change does not affect the connector builder or declarative connectors.

vercel · 2023-12-18T22:55:02Z

The latest updates on your projects. Learn more about Vercel for Git ↗︎

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	✅ Ready (Inspect)	Visit Preview	💬 Add feedback	Dec 19, 2023 11:57pm

bleonard · 2023-12-18T23:21:29Z

...airbyte-cdk/airbyte-commons/src/main/java/io/airbyte/commons/util/AutoCloseableIterator.java

+  default void close() throws Exception {
+    try {
+      closeSafely();
+    } catch (Exception e) {


Is there other stuff that would want to fail (doing clean up of some sort) in close where catching this doesn't make sense?

akashkulk · 2023-12-19T17:41:35Z

.../java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/base/IntegrationRunner.java

+  }
+
+  @VisibleForTesting
+  static void swallowIteratorCloseErrors(AutoCloseableIterator<AirbyteMessage> iter, Consumer<AirbyteMessage> outputRecordCollector) {


Is this called anywhere?

.../java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/base/IntegrationRunner.java

davinchia · 2023-12-19T21:06:50Z

@akashkulk PTAL! I removed all my WIP changes, so this should be much easier to review. Sorry and thanks!

This reverts commit ee02942.

This reverts commit bf6ef3e.

davinchia · 2023-12-19T23:54:28Z

.../java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/base/IntegrationRunner.java

            }
          } finally {
+            stopOrphanedThreads(EXIT_HOOK,


call this only once instead of twice in each method.

davinchia · 2023-12-19T23:55:59Z

@akashkulk @stephane-airbyte @rodireich PTAL - I think this is a pretty good middle ground to implementing the exception ignoring at the integration runner level while making things testable.

If this makes sense, I will add more tests!

davinchia · 2023-12-19T23:56:43Z

.../java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/base/IntegrationRunner.java

+              final Collection<AutoCloseableIterator<AirbyteMessage>> streams = source.readStreams(config, catalog, stateOptional.orElse(null));
+              final ConcurrentStreamConsumer streamConsumer = new ConcurrentStreamConsumer(this::consumeFromStream, streams.size());
+              readConcurrent(streams, streamConsumer);
            } else {
-              readSerial(config, catalog, stateOptional);
+              AutoCloseableIterator<AirbyteMessage> messageIterator = source.read(config, catalog, stateOptional.orElse(null));
+              readSerial(messageIterator, outputRecordCollector);


generally:

standardise the method signature.

make them static so they are easily tested/mocked in tests.

Checkpoint: tentative implementation.

ee02942

octavia-squidington-iii added the CDK Connector Development Kit label Dec 18, 2023

bleonard reviewed Dec 18, 2023

View reviewed changes

akashkulk reviewed Dec 19, 2023

View reviewed changes

.../java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/base/IntegrationRunner.java Outdated Show resolved Hide resolved

Remove WIP.

bf6ef3e

davinchia commented Dec 19, 2023

View reviewed changes

.../java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/base/IntegrationRunner.java Outdated Show resolved Hide resolved

davinchia added 2 commits December 19, 2023 13:05

Move things back.

ba9dd59

Remove bad import.

2913a66

davinchia added 5 commits December 19, 2023 14:57

Revert "Checkpoint: tentative implementation."

f7042f0

This reverts commit ee02942.

Revert "Remove WIP."

3e4aa52

This reverts commit bf6ef3e.

Checkpoint: implement for readSerial.

797d0f7

Better ignore exception on close.

6af53c5

Undo WIP.

e197441

davinchia commented Dec 19, 2023

View reviewed changes

Merge branch 'master' into davinchia/source-read-ignore-exception-close

0776977

davinchia commented Dec 19, 2023

View reviewed changes

vercel bot deployed to Preview December 19, 2023 23:57 View deployment

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

🐛 Do not propagate exceptions on source close. #33609

🐛 Do not propagate exceptions on source close. #33609

davinchia commented Dec 18, 2023 •

edited

Loading

vercel bot commented Dec 18, 2023 •

edited

Loading

bleonard Dec 18, 2023

akashkulk Dec 19, 2023

davinchia commented Dec 19, 2023

davinchia Dec 19, 2023

davinchia commented Dec 19, 2023

davinchia Dec 19, 2023

🐛 Do not propagate exceptions on source close. #33609

Are you sure you want to change the base?

🐛 Do not propagate exceptions on source close. #33609

Conversation

davinchia commented Dec 18, 2023 • edited Loading

What

How

Recommended reading order

🚨 User Impact 🚨

Pre-merge Actions

Community member or Airbyter

Airbyter

Community member or Airbyter

Airbyter

Airbyter

vercel bot commented Dec 18, 2023 • edited Loading

bleonard Dec 18, 2023

Choose a reason for hiding this comment

akashkulk Dec 19, 2023

Choose a reason for hiding this comment

davinchia commented Dec 19, 2023

davinchia Dec 19, 2023

Choose a reason for hiding this comment

davinchia commented Dec 19, 2023

davinchia Dec 19, 2023

Choose a reason for hiding this comment

davinchia commented Dec 18, 2023 •

edited

Loading

vercel bot commented Dec 18, 2023 •

edited

Loading