Merge branch 'master' into ddavydov/#20703-source-salesforce-fix-prop…

…s-chunk-length

airbyte-api/src/main/openapi/config.yaml

@@ -2789,6 +2789,8 @@ components:
           format: uuid
         disable_cache:
           type: boolean
+        notifySchemaChange:
+          type: boolean
     SourceUpdate:
       type: object
       required:

airbyte-bootloader/build.gradle

@@ -23,7 +23,7 @@ dependencies {
     implementation project(':airbyte-config:config-persistence')
     implementation project(':airbyte-db:db-lib')
     implementation project(":airbyte-json-validation")
-    implementation project(':airbyte-protocol:protocol-models')
+    implementation libs.airbyte.protocol
     implementation project(':airbyte-persistence:job-persistence')
 
     testAnnotationProcessor platform(libs.micronaut.bom)

airbyte-cdk/python/.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.28.0
+current_version = 0.28.1
 commit = False
 
 [bumpversion:file:setup.py]

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.28.1
+Low-Code CDK: fix signature _parse_records_and_emit_request_and_responses
+
 ## 0.28.0
 Low-Code: improve day_delta macro and MinMaxDatetime component
 

airbyte-cdk/python/README.md

@@ -71,7 +71,8 @@ If the iteration you are working on includes changes to the models, you might wa
 ```commandline
 SUB_BUILD=CONNECTORS_BASE ./gradlew format --scan --info --stacktrace
 ```
-This will generate the files based on the schemas, add the license information and format the code. If you want to only do the former and rely on pre-commit to the others, you can run the appropriate generation command i.e. `./gradlew generateProtocolClassFiles` or `./gradlew generateComponentManifestClassFiles`.
+This will generate the files based on the schemas, add the license information and format the code. If you want to only do the former and rely on
+pre-commit to the others, you can run the appropriate generation command i.e. `./gradlew generateComponentManifestClassFiles`.
 
 #### Testing
 

airbyte-cdk/python/airbyte_cdk/models/__init__.py

@@ -1,3 +1,9 @@
-# generated by generate-protocol-files
+# The earlier versions of airbyte-cdk (0.28.0<=) had the airbyte_protocol python classes
+# declared inline in the airbyte-cdk code. However, somewhere around Feb 2023 the
+# Airbyte Protocol moved to its own repo/PyPi package, called airbyte-protocol-models.
+# This directory including the airbyte_protocol.py and well_known_types.py files
+# are just wrappers on top of that stand-alone package which do some namespacing magic
+# to make the airbyte_protocol python classes available to the airbyte-cdk consumer as part
+# of airbyte-cdk rather than a standalone package.
 from .airbyte_protocol import *
 from .well_known_types import *

airbyte-cdk/python/airbyte_cdk/models/well_known_types.py

@@ -2,85 +2,4 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-# generated by datamodel-codegen:
-#   filename:  well_known_types.yaml
-
-from __future__ import annotations
-
-from enum import Enum
-from typing import Any, Union
-
-from pydantic import BaseModel, Field, constr
-
-
-class Model(BaseModel):
-    __root__: Any
-
-
-class String(BaseModel):
-    __root__: str = Field(..., description="Arbitrary text")
-
-
-class BinaryData(BaseModel):
-    __root__: constr(regex=r"^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$") = Field(
-        ...,
-        description="Arbitrary binary data. Represented as base64-encoded strings in the JSON transport. In the future, if we support other transports, may be encoded differently.\n",
-    )
-
-
-class Date(BaseModel):
-    __root__: constr(regex=r"^\d{4}-\d{2}-\d{2}( BC)?$") = Field(
-        ..., description="RFC 3339§5.6's full-date format, extended with BC era support"
-    )
-
-
-class TimestampWithTimezone(BaseModel):
-    __root__: constr(regex=r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+\-]\d{1,2}:\d{2})( BC)?$") = Field(
-        ...,
-        description='An instant in time. Frequently simply referred to as just a timestamp, or timestamptz. Uses RFC 3339§5.6\'s date-time format, requiring a "T" separator, and extended with BC era support. Note that we do _not_ accept Unix epochs here.\n',
-    )
-
-
-class TimestampWithoutTimezone(BaseModel):
-    __root__: constr(regex=r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?( BC)?$") = Field(
-        ...,
-        description='Also known as a localdatetime, or just datetime. Under RFC 3339§5.6, this would be represented as `full-date "T" partial-time`, extended with BC era support.\n',
-    )
-
-
-class TimeWithTimezone(BaseModel):
-    __root__: constr(regex=r"^\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+\-]\d{1,2}:\d{2})$") = Field(..., description="An RFC 3339§5.6 full-time")
-
-
-class TimeWithoutTimezone(BaseModel):
-    __root__: constr(regex=r"^\d{2}:\d{2}:\d{2}(\.\d+)?$") = Field(..., description="An RFC 3339§5.6 partial-time")
-
-
-class NumberEnum(Enum):
-    Infinity = "Infinity"
-    _Infinity = "-Infinity"
-    NaN = "NaN"
-
-
-class Number(BaseModel):
-    __root__: Union[Any, NumberEnum] = Field(
-        ...,
-        description="Note the mix of regex validation for normal numbers, and enum validation for special values.",
-    )
-
-
-class IntegerEnum(Enum):
-    Infinity = "Infinity"
-    _Infinity = "-Infinity"
-    NaN = "NaN"
-
-
-class Integer(BaseModel):
-    __root__: Union[Any, IntegerEnum]
-
-
-class Boolean(BaseModel):
-    __root__: bool = Field(
-        ...,
-        description="Note the direct usage of a primitive boolean rather than string. Unlike Numbers and Integers, we don't expect unusual values  here.",
-    )
+from airbyte_protocol.models.well_known_types import *

airbyte-cdk/python/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -407,7 +407,7 @@ def state(self, value: StreamState):
         """State setter, accept state serialized by state getter."""
         self.stream_slicer.update_cursor(value)
 
-    def _parse_records_and_emit_request_and_responses(self, request, response, stream_slice, stream_state) -> Iterable[StreamData]:
+    def _parse_records_and_emit_request_and_responses(self, request, response, stream_state, stream_slice) -> Iterable[StreamData]:
         # Only emit requests and responses when running in debug mode
         if self.logger.isEnabledFor(logging.DEBUG):
             yield _prepared_request_to_airbyte_message(request)

airbyte-cdk/python/build.gradle

@@ -7,12 +7,6 @@ airbytePython {
     moduleDirectory 'airbyte_cdk'
 }
 
-task generateProtocolClassFiles(type: Exec) {
-    environment 'ROOT_DIR', rootDir.absolutePath
-    commandLine 'bin/generate-protocol-files.sh'
-    dependsOn ':tools:code-generator:airbyteDocker'
-}
-
 task generateComponentManifestClassFiles(type: Exec) {
     environment 'ROOT_DIR', rootDir.absolutePath
     commandLine 'bin/generate-component-manifest-files.sh'
@@ -24,11 +18,6 @@ task validateSourceYamlManifest(type: Exec) {
     commandLine 'bin/validate-yaml-schema.sh'
 }
 
-blackFormat.dependsOn generateProtocolClassFiles
-isortFormat.dependsOn generateProtocolClassFiles
-flakeCheck.dependsOn generateProtocolClassFiles
-installReqs.dependsOn generateProtocolClassFiles
-
 blackFormat.dependsOn generateComponentManifestClassFiles
 isortFormat.dependsOn generateComponentManifestClassFiles
 flakeCheck.dependsOn generateComponentManifestClassFiles

airbyte-cdk/python/setup.py

@@ -15,7 +15,7 @@
 
 setup(
     name="airbyte-cdk",
-    version="0.28.0",
+    version="0.28.1",
     description="A framework for writing Airbyte Connectors.",
     long_description=README,
     long_description_content_type="text/markdown",
@@ -44,6 +44,7 @@
     packages=find_packages(exclude=("unit_tests",)),
     package_data={"airbyte_cdk": ["py.typed", "sources/declarative/declarative_component_schema.yaml"]},
     install_requires=[
+        "airbyte-protocol-models==1.0.0",
         "backoff",
         # pinned to the last working version for us temporarily while we fix
         "dataclasses-jsonschema==2.15.1",

airbyte-commons-protocol/build.gradle

@@ -5,7 +5,7 @@ dependencies {
     implementation libs.bundles.micronaut.annotation
     testImplementation libs.bundles.micronaut.test
 
-    implementation project(':airbyte-protocol:protocol-models')
+    implementation libs.airbyte.protocol
     implementation project(':airbyte-json-validation')
 }
 

airbyte-commons-server/build.gradle

@@ -39,7 +39,7 @@ dependencies {
     implementation project(':airbyte-db:db-lib')
     implementation project(":airbyte-json-validation")
     implementation project(':airbyte-oauth')
-    implementation project(':airbyte-protocol:protocol-models')
+    implementation libs.airbyte.protocol
     implementation project(':airbyte-persistence:job-persistence')
 
     implementation 'com.github.slugify:slugify:2.4'

airbyte-commons-server/src/main/java/io/airbyte/commons/server/handlers/SchedulerHandler.java

@@ -70,6 +70,7 @@
 import io.airbyte.config.persistence.SecretsRepositoryReader;
 import io.airbyte.config.persistence.SecretsRepositoryWriter;
 import io.airbyte.persistence.job.JobPersistence;
+import io.airbyte.persistence.job.WebUrlHelper;
 import io.airbyte.persistence.job.models.Job;
 import io.airbyte.protocol.models.AirbyteCatalog;
 import io.airbyte.protocol.models.ConnectorSpecification;
@@ -81,8 +82,10 @@
 import java.util.Optional;
 import java.util.UUID;
 import javax.validation.constraints.NotNull;
+import lombok.extern.slf4j.Slf4j;
 
 @Singleton
+@Slf4j
 public class SchedulerHandler {
 
   private static final HashFunction HASH_FUNCTION = Hashing.md5();
@@ -100,6 +103,7 @@ public class SchedulerHandler {
   private final JobConverter jobConverter;
   private final EventRunner eventRunner;
   private final FeatureFlags envVariableFeatureFlags;
+  private final WebUrlHelper webUrlHelper;
 
   // TODO: Convert to be fully using micronaut
   public SchedulerHandler(final ConfigRepository configRepository,
@@ -111,7 +115,8 @@ public SchedulerHandler(final ConfigRepository configRepository,
                           final LogConfigs logConfigs,
                           final EventRunner eventRunner,
                           final ConnectionsHandler connectionsHandler,
-                          final FeatureFlags envVariableFeatureFlags) {
+                          final FeatureFlags envVariableFeatureFlags,
+                          final WebUrlHelper webUrlHelper) {
     this(
         configRepository,
         secretsRepositoryWriter,
@@ -122,7 +127,8 @@ public SchedulerHandler(final ConfigRepository configRepository,
         eventRunner,
         new JobConverter(workerEnvironment, logConfigs),
         connectionsHandler,
-        envVariableFeatureFlags);
+        envVariableFeatureFlags,
+        webUrlHelper);
   }
 
   @VisibleForTesting
@@ -135,7 +141,8 @@ public SchedulerHandler(final ConfigRepository configRepository,
                    final EventRunner eventRunner,
                    final JobConverter jobConverter,
                    final ConnectionsHandler connectionsHandler,
-                   final FeatureFlags envVariableFeatureFlags) {
+                   final FeatureFlags envVariableFeatureFlags,
+                   final WebUrlHelper webUrlHelper) {
     this.configRepository = configRepository;
     this.secretsRepositoryWriter = secretsRepositoryWriter;
     this.synchronousSchedulerClient = synchronousSchedulerClient;
@@ -146,6 +153,7 @@ public SchedulerHandler(final ConfigRepository configRepository,
     this.jobConverter = jobConverter;
     this.connectionsHandler = connectionsHandler;
     this.envVariableFeatureFlags = envVariableFeatureFlags;
+    this.webUrlHelper = webUrlHelper;
   }
 
   public CheckConnectionRead checkSourceConnectionFromSourceId(final SourceIdRequestBody sourceIdRequestBody)
@@ -272,7 +280,7 @@ public SourceDiscoverSchemaRead discoverSchemaForSourceFromSourceId(final Source
 
       if (persistedCatalogId.isSuccess() && discoverSchemaRequestBody.getConnectionId() != null) {
         // modify discoveredSchema object to add CatalogDiff, containsBreakingChange, and connectionStatus
-        generateCatalogDiffsAndDisableConnectionsIfNeeded(discoveredSchema, discoverSchemaRequestBody);
+        generateCatalogDiffsAndDisableConnectionsIfNeeded(discoveredSchema, discoverSchemaRequestBody, source.getWorkspaceId());
       }
 
       return discoveredSchema;
@@ -394,9 +402,10 @@ public JobInfoRead cancelJob(final JobIdRequestBody jobIdRequestBody) throws IOE
   // determine whether 1. the source schema change resulted in a broken connection or 2. the user
   // wants the connection disabled when non-breaking changes are detected. If so, disable that
   // connection. Modify the current discoveredSchema object to add a CatalogDiff,
-  // containsBreakingChange paramter, and connectionStatus parameter.
+  // containsBreakingChange parameter, and connectionStatus parameter.
   private void generateCatalogDiffsAndDisableConnectionsIfNeeded(final SourceDiscoverSchemaRead discoveredSchema,
-                                                                 final SourceDiscoverSchemaRequestBody discoverSchemaRequestBody)
+                                                                 final SourceDiscoverSchemaRequestBody discoverSchemaRequestBody,
+                                                                 final UUID workspaceId)
       throws JsonValidationException, ConfigNotFoundException, IOException {
     final ConnectionReadList connectionsForSource = connectionsHandler.listConnectionsForSource(discoverSchemaRequestBody.getSourceId(), false);
     for (final ConnectionRead connectionRead : connectionsForSource.getConnections()) {
@@ -418,12 +427,23 @@ private void generateCatalogDiffsAndDisableConnectionsIfNeeded(final SourceDisco
       }
       updateObject.status(connectionStatus);
       connectionsHandler.updateConnection(updateObject);
+      if (shouldNotifySchemaChange(diff, connectionRead, discoverSchemaRequestBody)) {
+        final String url = webUrlHelper.getConnectionUrl(workspaceId, connectionRead.getConnectionId());
+        eventRunner.sendSchemaChangeNotification(connectionRead.getConnectionId(), url);
+      }
       if (connectionRead.getConnectionId().equals(discoverSchemaRequestBody.getConnectionId())) {
         discoveredSchema.catalogDiff(diff).breakingChange(containsBreakingChange).connectionStatus(connectionStatus);
       }
     }
   }
 
+  private boolean shouldNotifySchemaChange(final CatalogDiff diff,
+                                           final ConnectionRead connectionRead,
+                                           final SourceDiscoverSchemaRequestBody requestBody) {
+    return !diff.getTransforms().isEmpty() && connectionRead.getNotifySchemaChanges() && requestBody.getNotifySchemaChange() != null
+        && requestBody.getNotifySchemaChange();
+  }
+
   private boolean shouldDisableConnection(final boolean containsBreakingChange,
                                           final NonBreakingChangesPreference preference,
                                           final CatalogDiff diff) {

airbyte-commons-server/src/main/java/io/airbyte/commons/server/handlers/WebBackendConnectionsHandler.java

@@ -413,7 +413,8 @@ private Optional<SourceDiscoverSchemaRead> getRefreshedSchema(final UUID sourceI
     final SourceDiscoverSchemaRequestBody discoverSchemaReadReq = new SourceDiscoverSchemaRequestBody()
         .sourceId(sourceId)
         .disableCache(true)
-        .connectionId(connectionId);
+        .connectionId(connectionId)
+        .notifySchemaChange(false);
     final SourceDiscoverSchemaRead schemaRead = schedulerHandler.discoverSchemaForSourceFromSourceId(discoverSchemaReadReq);
     return Optional.ofNullable(schemaRead);
   }

airbyte-commons-server/src/main/java/io/airbyte/commons/server/scheduler/EventRunner.java

@@ -28,4 +28,6 @@ public interface EventRunner {
 
   void update(final UUID connectionId);
 
+  void sendSchemaChangeNotification(final UUID connectionId, final String url);
+
 }

airbyte-commons-server/src/main/java/io/airbyte/commons/server/scheduler/TemporalEventRunner.java

@@ -54,4 +54,9 @@ public void update(final UUID connectionId) {
     temporalClient.update(connectionId);
   }
 
+  @Override
+  public void sendSchemaChangeNotification(final UUID connectionId, final String url) {
+    temporalClient.sendSchemaChangeNotification(connectionId, url);
+  }
+
 }