Merge branch 'master' into 13758-SSH_control_host_port_setting_not_used_when_tunneling

Author: kimerinn

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.1.69
+- AbstractSource emits a state message when reading incremental even if there were no stream slices to process.
+
 ## 0.1.68
 - Replace parse-time string interpolation with run-time interpolation in YAML-based sources
 

airbyte-cdk/python/airbyte_cdk/sources/abstract_source.py

@@ -226,6 +226,10 @@ def _read_incremental(
         )
         logger.debug(f"Processing stream slices for {stream_name}", extra={"stream_slices": slices})
         total_records_counter = 0
+        if not slices:
+            # Safety net to ensure we always emit at least one state message even if there are no slices
+            checkpoint = self._checkpoint_state(stream_instance, stream_instance.state, connector_state)
+            yield checkpoint
         for _slice in slices:
             logger.debug("Processing stream slice", extra={"slice": _slice})
             records = stream_instance.read_records(

airbyte-cdk/python/setup.py

@@ -15,7 +15,7 @@
 
 setup(
     name="airbyte-cdk",
-    version="0.1.68",
+    version="0.1.69",
     description="A framework for writing Airbyte Connectors.",
     long_description=README,
     long_description_content_type="text/markdown",

airbyte-cdk/python/unit_tests/sources/test_abstract_source.py

@@ -98,9 +98,13 @@ def primary_key(self) -> Optional[Union[str, List[str], List[List[str]]]]:
 class MockStreamWithState(MockStream):
     cursor_field = "cursor"
 
+    def __init__(self, inputs_and_mocked_outputs: List[Tuple[Mapping[str, Any], Iterable[Mapping[str, Any]]]], name: str, state=None):
+        super().__init__(inputs_and_mocked_outputs, name)
+        self._state = state
+
     @property
     def state(self):
-        return {}
+        return self._state
 
     @state.setter
     def state(self, value):
@@ -452,6 +456,74 @@ def test_with_slices(self, mocker):
 
         assert expected == messages
 
+    def test_no_slices(self, mocker):
+        """
+        Tests that an incremental read returns at least one state messages even if no records were read:
+            1. outputs a state message after reading the entire stream
+        """
+        slices = []
+        stream_output = [{"k1": "v1"}, {"k2": "v2"}, {"k3": "v3"}]
+        state = {"cursor": "value"}
+        s1 = MockStreamWithState(
+            [
+                (
+                    {
+                        "sync_mode": SyncMode.incremental,
+                        "stream_slice": s,
+                        "stream_state": mocker.ANY,
+                    },
+                    stream_output,
+                )
+                for s in slices
+            ],
+            name="s1",
+            state=state,
+        )
+        s2 = MockStreamWithState(
+            [
+                (
+                    {
+                        "sync_mode": SyncMode.incremental,
+                        "stream_slice": s,
+                        "stream_state": mocker.ANY,
+                    },
+                    stream_output,
+                )
+                for s in slices
+            ],
+            name="s2",
+            state=state,
+        )
+
+        mocker.patch.object(MockStreamWithState, "supports_incremental", return_value=True)
+        mocker.patch.object(MockStreamWithState, "get_json_schema", return_value={})
+        mocker.patch.object(MockStreamWithState, "stream_slices", return_value=slices)
+        mocker.patch.object(
+            MockStreamWithState,
+            "state_checkpoint_interval",
+            new_callable=mocker.PropertyMock,
+            return_value=2,
+        )
+
+        src = MockSource(streams=[s1, s2])
+        catalog = ConfiguredAirbyteCatalog(
+            streams=[
+                _configured_stream(s1, SyncMode.incremental),
+                _configured_stream(s2, SyncMode.incremental),
+            ]
+        )
+
+        expected = [
+            _state({"s1": state}),
+            _state({"s1": state, "s2": state}),
+        ]
+
+        messages = _fix_emitted_at(list(src.read(logger, {}, catalog, state=defaultdict(dict))))
+
+        print(f"expected:\n{expected}")
+        print(f"messages:\n{messages}")
+        assert expected == messages
+
     def test_with_slices_and_interval(self, mocker):
         """
         Tests that an incremental read which uses slices and a checkpoint interval:

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -53,14 +53,14 @@
 - name: Cassandra
   destinationDefinitionId: 707456df-6f4f-4ced-b5c6-03f73bcad1c5
   dockerRepository: airbyte/destination-cassandra
-  dockerImageTag: 0.1.2
+  dockerImageTag: 0.1.3
   documentationUrl: https://docs.airbyte.io/integrations/destinations/cassandra
   icon: cassandra.svg
   releaseStage: alpha
 - name: Chargify (Keen)
   destinationDefinitionId: 81740ce8-d764-4ea7-94df-16bb41de36ae
   dockerRepository: airbyte/destination-keen
-  dockerImageTag: 0.2.3
+  dockerImageTag: 0.2.4
   documentationUrl: https://docs.airbyte.io/integrations/destinations/keen
   icon: chargify.svg
   releaseStage: alpha
@@ -133,7 +133,7 @@
 - name: Kafka
   destinationDefinitionId: 9f760101-60ae-462f-9ee6-b7a9dafd454d
   dockerRepository: airbyte/destination-kafka
-  dockerImageTag: 0.1.9
+  dockerImageTag: 0.1.10
   documentationUrl: https://docs.airbyte.io/integrations/destinations/kafka
   icon: kafka.svg
   releaseStage: alpha

airbyte-config/init/src/main/resources/seed/destination_specs.yaml

@@ -693,7 +693,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-cassandra:0.1.2"
+- dockerImage: "airbyte/destination-cassandra:0.1.3"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/cassandra"
     connectionSpecification:
@@ -758,7 +758,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-keen:0.2.3"
+- dockerImage: "airbyte/destination-keen:0.2.4"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/keen"
     connectionSpecification:
@@ -2005,7 +2005,7 @@
     supportsDBT: false
     supported_destination_sync_modes:
     - "append"
-- dockerImage: "airbyte/destination-kafka:0.1.9"
+- dockerImage: "airbyte/destination-kafka:0.1.10"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/kafka"
     connectionSpecification:

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -564,7 +564,7 @@
 - name: Microsoft SQL Server (MSSQL)
   sourceDefinitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
   dockerRepository: airbyte/source-mssql
-  dockerImageTag: 0.4.12
+  dockerImageTag: 0.4.13
   documentationUrl: https://docs.airbyte.io/integrations/sources/mssql
   icon: mssql.svg
   sourceType: database

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -5003,7 +5003,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-mssql:0.4.12"
+- dockerImage: "airbyte/source-mssql:0.4.13"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/mssql"
     connectionSpecification:

airbyte-integrations/connectors/destination-cassandra/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION destination-cassandra
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.2
+LABEL io.airbyte.version=0.1.3
 LABEL io.airbyte.name=airbyte/destination-cassandra

airbyte-integrations/connectors/destination-cassandra/build.gradle

@@ -26,6 +26,7 @@ dependencies {
     // https://mvnrepository.com/artifact/org.assertj/assertj-core
     testImplementation "org.assertj:assertj-core:${assertVersion}"
     testImplementation libs.connectors.testcontainers.cassandra
+    testImplementation project(':airbyte-integrations:bases:standard-destination-test')
 
 
     integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')