🎉 Source Amazon SQS: New connector #6937

sdairs · 2021-10-10T18:22:15Z

What

Adds an Amazon SQS Source connector
Has Unit tests for read/check/discover with mock AWS resources to simulate behaviour without needing to create a real SQS topic or IAM roles

airbytehq/connector-contest#68

How

Connector built with Python CDK
Uses python boto3 for AWS functionality
Uses python moto for mock-AWS resources

Pre-merge Checklist

Expand the relevant checklist and delete the others.

New Connector

Community member or Airbyter

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
Credentials added to Github CI. Instructions.
/test connector=connectors/<name> command is passing.
New Connector version released on Dockerhub by running the /publish command described here

Updating a connector

Community member or Airbyter

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
Credentials added to Github CI. Instructions.
/test connector=connectors/<name> command is passing.
New Connector version released on Dockerhub by running the /publish command described here

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:testScaffoldTemplates then checking in your changes
Documentation which references the generator is updated as needed.

CLAassistant · 2021-10-10T18:22:21Z

All committers have signed the CLA.

sdairs · 2021-10-10T18:27:46Z

Right now test_read integration test fails as there are no records to verify in the stream.

In the unit tests I am using moto to create mock AWS services, but I'm not sure how this works with the integration/acceptance testing method. All other standard tests pass.

collecting ...
 test_core.py ✓✓✓✓✓✓✓✓✓✓                                                                                                                                                                                                                        83% ████████▍

―――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――― TestBasicRead.test_read[inputs0] ――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――

self = <source_acceptance_test.tests.test_core.TestBasicRead object at 0x7f39cc406450>, connector_config = SecretDict(******)
configured_catalog = ConfiguredAirbyteCatalog(streams=[ConfiguredAirbyteStream(stream=AirbyteStream(name='ab-airbyte-testing', json_schema=...: 'full_refresh'>, cursor_field=None, destination_sync_mode=<DestinationSyncMode.append: 'append'>, primary_key=None)])
inputs = BasicReadTestConfig(config_path='secrets/config.json', configured_catalog_path='integration_tests/configured_catalog.json', empty_streams=set(), expect_records=None, validate_schema=True, timeout_seconds=None), expected_records = []
docker_runner = <source_acceptance_test.utils.connector_runner.ConnectorRunner object at 0x7f39cc406e50>, detailed_logger = <Logger detailed_logger /test_input/acceptance_tests_logs/test_core.py__TestBasicRead__test_read[inputs0].txt (DEBUG)>

    def test_read(
        self,
        connector_config,
        configured_catalog,
        inputs: BasicReadTestConfig,
        expected_records: List[AirbyteMessage],
        docker_runner: ConnectorRunner,
        detailed_logger,
    ):
        output = docker_runner.call_read(connector_config, configured_catalog)
        records = [message.record for message in filter_output(output, Type.RECORD)]

>       assert records, "At least one record should be read using provided catalog"
E       AssertionError: At least one record should be read using provided catalog
E       assert []

source_acceptance_test/tests/test_core.py:259: AssertionError

 test_core.py ⨯                                                                                                                                                                                                                                 92% █████████▎



 test_full_refresh.py ✓                                                                                                                                                                                                                        100% ██████████
================================================================================================================== short test summary info ===================================================================================================================
FAILED ../../test_input/test_core.py::TestBasicRead::test_read[inputs0] - AssertionError: At least one record should be read using provided catalog

Results (71.40s):
      11 passed
       1 failed
         - ../airbyte/source_acceptance_test/source_acceptance_test/tests/test_core.py:247 TestBasicRead.test_read[inputs0]

sdairs · 2021-10-14T22:30:26Z

Have managed to get a successful run of the acceptance & units tests - output below

> Task :airbyte-integrations:connectors:source-amazon-sqs:unitTest
         unit_tests/unit_test.py::test_read {"type": "LOG", "log": {"level": "DEBUG", "message": "Amazon SQS Source Read - stream is: amazon-sqs-mock-queue"}}
         {"type": "LOG", "log": {"level": "DEBUG", "message": "Amazon SQS Source Read - Creating SQS connection ---"}}
         {"type": "LOG", "log": {"level": "DEBUG", "message": "Amazon SQS Source Read - Connected to SQS Queue ---"}}
         {"type": "LOG", "log": {"level": "DEBUG", "message": "Amazon SQS Source Read - Beginning message poll ---"}}
         {"type": "LOG", "log": {"level": "DEBUG", "message": "Amazon SQS Source Read - Message recieved: c9e561e9-9907-84d6-17ad-9557daf34efd"}}
         {"type": "LOG", "log": {"level": "DEBUG", "message": "Amazon SQS Source Read - Setting message visibility timeout: c9e561e9-9907-84d6-17ad-9557daf34efd"}}
         {"type": "LOG", "log": {"level": "DEBUG", "message": "Amazon SQS Source Read - Message visibility timeout set: c9e561e9-9907-84d6-17ad-9557daf34efd"}}
         {"type": "LOG", "log": {"level": "DEBUG", "message": "Amazon SQS Source Read - Beginning message poll ---"}}
         {"type": "LOG", "log": {"level": "DEBUG", "message": "Amazon SQS Source Read - No messages recieved during poll, time out reached ---"}}
         PASSED

         ----------- coverage: platform linux, python 3.9.7-final-0 -----------
         Name                            Stmts   Miss  Cover
         ---------------------------------------------------
         source_amazon_sqs/__init__.py       2      0   100%
         source_amazon_sqs/source.py        99     20    80%
         ---------------------------------------------------
         TOTAL                             101     20    80%
         Coverage HTML written to dir htmlcov


         ============================== 3 passed in 6.24s ===============================

Successfully tagged airbyte/source-amazon-sqs:dev

> Task :airbyte-integrations:connectors:source-amazon-sqs:sourceAcceptanceTest
============================= test session starts ==============================
platform linux -- Python 3.7.11, pytest-6.2.5, py-1.10.0, pluggy-1.0.0
rootdir: /test_input
plugins: sugar-0.9.4, timeout-1.4.2
collected 12 items / 1 skipped / 11 selected

test_core.py ...........                                                 [ 91%]
test_full_refresh.py .                                                   [100%]

=================== 12 passed, 1 skipped in 71.25s (0:01:11) ===================

Deprecated Gradle features were used in this build, making it incompatible with Gradle 7.0.
Use '--warning-mode all' to show the individual deprecation warnings.
See https://docs.gradle.org/6.7.1/userguide/command_line_interface.html#sec:command_line_warnings

BUILD SUCCESSFUL in 2m 11s
80 actionable tasks: 46 executed, 34 up-to-date```

marcosmarxm · 2021-10-19T17:47:29Z

awesome @sdairs I'm going to review and test your connector later today.

marcosmarxm · 2021-10-20T14:47:15Z

airbyte-integrations/connectors/source-amazon-sqs/bootstrap.md

+### Deletes
+Optionally, it can delete messages after reading - the delete_message() call is made __after__ yielding the message to the generator.
+This means that messages aren't deleted unless read by a Destination - however, there is still potential that this could result in 
+missed messages if the Destination fails __after__ taking the message, but before commiting to to its own downstream.


@sherifnada can you review this feature from Amazon SQS connector. Make sense enable this to an Airbyte Connector?

I'm hesitant to do this exactly for the reason mentioned. I would suggest we skip deleting messages. But my question then is: how do we do incremental sync? Is it possible to only read new messages?

It would not be possible to do incremental sync without delete as there is no concept of offsets in SQS - I think it's an important feature to have, but we should make the caveats clear with warnings & further details in documentation.

marcosmarxm

Unit and Integration Tests are running with CI Sandbox account. I'll wait @sherifnada feedback about the delete feature.

Can you move the boostrap.md to the connector doc page? Makes more sense when people are trying to setting up the connector already read this info.

sdairs · 2021-10-20T14:54:09Z

Awesome thanks for reviewing.

The delete message is a standard SQS feature that I think should be supported - it is false by default to avoid accidental deletes, but I would expect most use cases for SQS would want deletes enabled.

sdairs · 2021-10-20T15:01:46Z

There is of course potential that messages could be lost with Delete enabled - in that, if the Destination consumes the message, and then fails before commiting it to whatever is downstream, the Source will have deleted it after yielding it - in my limited understand of Airbyte so far, there is no persistence of messages that are 'in flight' i.e. being processed by a Destination or Source - so there would be no way to cover this message and have the Destination resume from where it left off.

A more robust method would be some sort of callback handle that a Destination could call after committing the message downstream, to notify the Source that the message is persisted and can be deleted. But I don't know if this is something that is possible - and would have performance considerations.

airbyte-integrations/connectors/source-amazon-sqs/bootstrap.md

sherifnada · 2021-10-21T05:01:10Z

airbyte-integrations/connectors/source-amazon-sqs/bootstrap.md

+### Deletes
+Optionally, it can delete messages after reading - the delete_message() call is made __after__ yielding the message to the generator.
+This means that messages aren't deleted unless read by a Destination - however, there is still potential that this could result in 
+missed messages if the Destination fails __after__ taking the message, but before commiting to to its own downstream.


I'm hesitant to do this exactly for the reason mentioned. I would suggest we skip deleting messages. But my question then is: how do we do incremental sync? Is it possible to only read new messages?

airbyte-integrations/connectors/source-amazon-sqs/source_amazon_sqs/spec.json

sdairs · 2021-10-25T16:18:00Z

Have added explicit warnings to Spec and further details to Readme to cover the delete option & potential data loss scenarios

marcosmarxm

LGTM thanks for your contribution @sdairs !

* Initial commit, working source with static Creds * Typo in example queue url * Adds auto delete of messages after read * Adds visibility timeout * remove insecure comments from AWS IAM Key spec * explicitly set supported sync modes * explicit sync mode should be lower case * Adds unit tests for check, discover, read * remove incremental acceptance test block * remove incremental from conf catalog sample * remove test requirement moto from main req * align int catalog sample with sample_files * fixing catalog configs * acceptance testing config * adds expected records txt * automated formatting changes * remove expected records block from acpt test * Adds Docs page * Ammends formatting on readme * Adds doc link to summary * Improve error handling & debug logging * Adds bootstrap.md * Add a todo suggestion for batch output * Adds SQS to integrations readme list * lower case properties * removed unused line * uses enum for aws region * updates sample configs to use lowercase * required props to lower case * add missed property to lowercase * gradle formatting * Fixing issues from acceptance tests * annotate secrets in spec.json with airbyte_secret * Adds explicit warnings about data less when using Delete Message option

sdairs added 20 commits October 9, 2021 17:52

Initial commit, working source with static Creds

9a7caf4

Typo in example queue url

4a79ce7

Adds auto delete of messages after read

137bf3c

Adds visibility timeout

92e1c93

remove insecure comments from AWS IAM Key spec

e11f936

explicitly set supported sync modes

2780160

explicit sync mode should be lower case

9eac21a

Adds unit tests for check, discover, read

932e376

remove incremental acceptance test block

1ebc016

remove incremental from conf catalog sample

d093456

remove test requirement moto from main req

1c0cab4

align int catalog sample with sample_files

6971564

fixing catalog configs

7cae0d9

acceptance testing config

a0b2c95

adds expected records txt

d0dec92

automated formatting changes

36f64ef

remove expected records block from acpt test

c3f7ed1

Adds Docs page

f1c8193

Ammends formatting on readme

3ae20c7

Adds doc link to summary

68d34ad

github-actions bot added area/connectors Connector related issues area/documentation Improvements or additions to documentation labels Oct 10, 2021

avaidyanatha added the community_pr label Oct 11, 2021

sdairs added 4 commits October 11, 2021 21:31

Improve error handling & debug logging

d637b51

Adds bootstrap.md

91e48c8

Add a todo suggestion for batch output

e99bb40

Adds SQS to integrations readme list

734bf52

marcosmarxm added the community label Oct 12, 2021

sdairs added 2 commits October 13, 2021 16:34

lower case properties

8994d3f

removed unused line

06f15b4

marcosmarxm mentioned this pull request Oct 13, 2021

CI Sandbox for Amazon SQS #7006

Closed

sdairs added 5 commits October 13, 2021 16:53

uses enum for aws region

6eea384

updates sample configs to use lowercase

8686364

required props to lower case

e0a1b8f

add missed property to lowercase

2c9c4d1

gradle formatting

63591cf

marcosmarxm removed the community_pr label Oct 14, 2021

sdairs requested a review from marcosmarxm October 14, 2021 15:23

Fixing issues from acceptance tests

bb48c8b

sdairs mentioned this pull request Oct 15, 2021

[kubernetes/helm] Allow authentication to AWS using roles #5942

Closed

annotate secrets in spec.json with airbyte_secret

57e735b

marcosmarxm reviewed Oct 20, 2021

View reviewed changes

marcosmarxm requested changes Oct 20, 2021

View reviewed changes

sherifnada reviewed Oct 21, 2021

View reviewed changes

Adds explicit warnings about data less when using Delete Message option

752a953

sdairs requested review from sherifnada and marcosmarxm October 25, 2021 16:18

marcosmarxm approved these changes Oct 25, 2021

View reviewed changes

marcosmarxm merged commit 58b569d into airbytehq:master Oct 26, 2021

jrhizor mentioned this pull request Nov 1, 2021

Bump Airbyte version from 0.30.23-alpha to 0.30.24-alpha #7532

Merged

marcosmarxm removed the hacktober label Oct 20, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

🎉 Source Amazon SQS: New connector #6937

🎉 Source Amazon SQS: New connector #6937

sdairs commented Oct 10, 2021 •

edited

CLAassistant commented Oct 10, 2021 •

edited

sdairs commented Oct 10, 2021

sdairs commented Oct 14, 2021 •

edited

marcosmarxm commented Oct 19, 2021

marcosmarxm Oct 20, 2021

sherifnada Oct 21, 2021

sdairs Oct 21, 2021

marcosmarxm left a comment

sdairs commented Oct 20, 2021

sdairs commented Oct 20, 2021

sherifnada Oct 21, 2021

sdairs commented Oct 25, 2021

marcosmarxm left a comment

🎉 Source Amazon SQS: New connector #6937

🎉 Source Amazon SQS: New connector #6937

Conversation

sdairs commented Oct 10, 2021 • edited

What

How

Recommended reading order

Pre-merge Checklist

Community member or Airbyter

Airbyter

Community member or Airbyter

Airbyter

CLAassistant commented Oct 10, 2021 • edited

sdairs commented Oct 10, 2021

sdairs commented Oct 14, 2021 • edited

marcosmarxm commented Oct 19, 2021

marcosmarxm Oct 20, 2021

Choose a reason for hiding this comment

sherifnada Oct 21, 2021

Choose a reason for hiding this comment

sdairs Oct 21, 2021

Choose a reason for hiding this comment

marcosmarxm left a comment

Choose a reason for hiding this comment

sdairs commented Oct 20, 2021

sdairs commented Oct 20, 2021

sherifnada Oct 21, 2021

Choose a reason for hiding this comment

sdairs commented Oct 25, 2021

marcosmarxm left a comment

Choose a reason for hiding this comment

sdairs commented Oct 10, 2021 •

edited

CLAassistant commented Oct 10, 2021 •

edited

sdairs commented Oct 14, 2021 •

edited