Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Disable broken connections when feature flag is on #19730

Merged
merged 10 commits into from
Nov 29, 2022
Merged

Disable broken connections when feature flag is on #19730

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
a719466
remove from orchestrator
alovew Nov 15, 2022
c2ab26b
tests
alovew Nov 15, 2022
09e97e6
Revert "remove from orchestrator"
alovew Nov 16, 2022
7eb5826
revert
alovew Nov 16, 2022
7d2b98d
add connection status to sourceDiscoverSchemaRead
alovew Nov 22, 2022
11b7fc2
check connection preferences
alovew Nov 28, 2022
bc3aa28
tests
alovew Nov 28, 2022
e0c6fd4
check for diff
alovew Nov 28, 2022
59c6396
tests
alovew Nov 28, 2022
c5be1b0
inject env variable feature flags
alovew Nov 28, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions airbyte-api/src/main/openapi/config.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2935,6 +2935,8 @@ components:
$ref: "#/components/schemas/CatalogDiff"
breakingChange:
type: boolean
connectionStatus:
$ref: "#/components/schemas/ConnectionStatus"
SourceSearch:
type: object
properties:
Expand Down
6 changes: 5 additions & 1 deletion airbyte-server/src/main/java/io/airbyte/server/ServerApp.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import io.airbyte.analytics.Deployment;
import io.airbyte.analytics.TrackingClient;
import io.airbyte.analytics.TrackingClientSingleton;
import io.airbyte.commons.features.EnvVariableFeatureFlags;
import io.airbyte.commons.lang.CloseableShutdownHook;
import io.airbyte.commons.resources.MoreResources;
import io.airbyte.commons.temporal.ConnectionManagerUtils;
Expand Down Expand Up @@ -213,6 +214,8 @@ public static ServerRunnable getServer(final ServerFactory apiFactory,
final TrackingClient trackingClient = TrackingClientSingleton.get();
final JobTracker jobTracker = new JobTracker(configRepository, jobPersistence, trackingClient);

final EnvVariableFeatureFlags envVariableFeatureFlags = new EnvVariableFeatureFlags();

final WebUrlHelper webUrlHelper = new WebUrlHelper(configs.getWebappUrl());
final JobErrorReportingClient jobErrorReportingClient = JobErrorReportingClientFactory.getClient(configs.getJobErrorReportingStrategy(), configs);
final JobErrorReporter jobErrorReporter =
Expand Down Expand Up @@ -286,7 +289,8 @@ public static ServerRunnable getServer(final ServerFactory apiFactory,
configs.getWorkerEnvironment(),
configs.getLogConfigs(),
eventRunner,
connectionsHandler);
connectionsHandler,
envVariableFeatureFlags);

final DbMigrationHandler dbMigrationHandler = new DbMigrationHandler(configsDatabase, configsFlyway, jobsDatabase, jobsFlyway);

Expand Down
35 changes: 30 additions & 5 deletions airbyte-server/src/main/java/io/airbyte/server/handlers/SchedulerHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
import io.airbyte.api.model.generated.CheckConnectionRead;
import io.airbyte.api.model.generated.CheckConnectionRead.StatusEnum;
import io.airbyte.api.model.generated.ConnectionIdRequestBody;
import io.airbyte.api.model.generated.ConnectionRead;
import io.airbyte.api.model.generated.ConnectionStatus;
import io.airbyte.api.model.generated.ConnectionUpdate;
import io.airbyte.api.model.generated.DestinationCoreConfig;
import io.airbyte.api.model.generated.DestinationDefinitionIdWithWorkspaceId;
Expand All @@ -27,6 +29,7 @@
import io.airbyte.api.model.generated.JobIdRequestBody;
import io.airbyte.api.model.generated.JobInfoRead;
import io.airbyte.api.model.generated.LogRead;
import io.airbyte.api.model.generated.NonBreakingChangesPreference;
import io.airbyte.api.model.generated.SourceCoreConfig;
import io.airbyte.api.model.generated.SourceDefinitionIdWithWorkspaceId;
import io.airbyte.api.model.generated.SourceDefinitionSpecificationRead;
Expand All @@ -39,6 +42,7 @@
import io.airbyte.api.model.generated.SynchronousJobRead;
import io.airbyte.commons.docker.DockerUtils;
import io.airbyte.commons.enums.Enums;
import io.airbyte.commons.features.EnvVariableFeatureFlags;
import io.airbyte.commons.json.Jsons;
import io.airbyte.commons.temporal.ErrorCode;
import io.airbyte.commons.temporal.TemporalClient.ManualOperationResult;
Expand Down Expand Up @@ -95,6 +99,7 @@ public class SchedulerHandler {
private final JobPersistence jobPersistence;
private final JobConverter jobConverter;
private final EventRunner eventRunner;
private final EnvVariableFeatureFlags envVariableFeatureFlags;

public SchedulerHandler(final ConfigRepository configRepository,
final SecretsRepositoryReader secretsRepositoryReader,
Expand All @@ -104,7 +109,8 @@ public SchedulerHandler(final ConfigRepository configRepository,
final WorkerEnvironment workerEnvironment,
final LogConfigs logConfigs,
final EventRunner eventRunner,
final ConnectionsHandler connectionsHandler) {
final ConnectionsHandler connectionsHandler,
final EnvVariableFeatureFlags envVariableFeatureFlags) {
this(
configRepository,
secretsRepositoryWriter,
Expand All @@ -114,7 +120,8 @@ public SchedulerHandler(final ConfigRepository configRepository,
jobPersistence,
eventRunner,
new JobConverter(workerEnvironment, logConfigs),
connectionsHandler);
connectionsHandler,
envVariableFeatureFlags);
}

@VisibleForTesting
Expand All @@ -126,7 +133,8 @@ public SchedulerHandler(final ConfigRepository configRepository,
final JobPersistence jobPersistence,
final EventRunner eventRunner,
final JobConverter jobConverter,
final ConnectionsHandler connectionsHandler) {
final ConnectionsHandler connectionsHandler,
final EnvVariableFeatureFlags envVariableFeatureFlags) {
this.configRepository = configRepository;
this.secretsRepositoryWriter = secretsRepositoryWriter;
this.synchronousSchedulerClient = synchronousSchedulerClient;
Expand All @@ -136,6 +144,7 @@ public SchedulerHandler(final ConfigRepository configRepository,
this.eventRunner = eventRunner;
this.jobConverter = jobConverter;
this.connectionsHandler = connectionsHandler;
this.envVariableFeatureFlags = envVariableFeatureFlags;
}

public CheckConnectionRead checkSourceConnectionFromSourceId(final SourceIdRequestBody sourceIdRequestBody)
Expand Down Expand Up @@ -360,16 +369,32 @@ private void discoveredSchemaWithCatalogDiff(SourceDiscoverSchemaRead discovered
throws JsonValidationException, ConfigNotFoundException, IOException {
final Optional<io.airbyte.api.model.generated.AirbyteCatalog> catalogUsedToMakeConfiguredCatalog = connectionsHandler
.getConnectionAirbyteCatalog(discoverSchemaRequestBody.getConnectionId());
final ConnectionRead connectionRead = connectionsHandler.getConnection(discoverSchemaRequestBody.getConnectionId());
final io.airbyte.api.model.generated.@NotNull AirbyteCatalog currentAirbyteCatalog =
connectionsHandler.getConnection(discoverSchemaRequestBody.getConnectionId()).getSyncCatalog();
connectionRead.getSyncCatalog();
CatalogDiff diff = connectionsHandler.getDiff(catalogUsedToMakeConfiguredCatalog.orElse(currentAirbyteCatalog), discoveredSchema.getCatalog(),
CatalogConverter.toProtocol(currentAirbyteCatalog));
boolean containsBreakingChange = containsBreakingChange(diff);
ConnectionUpdate updateObject =
new ConnectionUpdate().breakingChange(containsBreakingChange).connectionId(discoverSchemaRequestBody.getConnectionId());
ConnectionStatus connectionStatus;
if (shouldDisableConnection(containsBreakingChange, connectionRead.getNonBreakingChangesPreference(), diff)) {
connectionStatus = ConnectionStatus.INACTIVE;
} else {
connectionStatus = ConnectionStatus.ACTIVE;
}
updateObject.status(connectionStatus);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need a test that check that we are setting the expected status.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have a bunch of new SchedulerHandlerTests -- is there another case you think I should be testing?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The status is being tested on each new test, and I also added it to existing tests

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a test that check that we are not disabling the connection if the feature flag is set to false or if the preference is set to not disable?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes - you have to expand SchedulerHandlerTest in github to see it because I think it is collapsed due to the large number of changes to the file

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh my bad, I missed them because of them being collapsed 🤦

connectionsHandler.updateConnection(updateObject);
discoveredSchema.catalogDiff(diff).breakingChange(containsBreakingChange);
discoveredSchema.catalogDiff(diff).breakingChange(containsBreakingChange).connectionStatus(connectionStatus);

}

private boolean shouldDisableConnection(boolean containsBreakingChange, NonBreakingChangesPreference preference, CatalogDiff diff) {
if (!envVariableFeatureFlags.autoDetectSchema()) {
return false;
}

return containsBreakingChange || (preference == NonBreakingChangesPreference.DISABLE && !diff.getTransforms().isEmpty());
}

private CheckConnectionRead reportConnectionStatus(final SynchronousResponse<StandardCheckConnectionOutput> response) {
Expand Down
1 change: 1 addition & 0 deletions airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -353,6 +353,7 @@ public WebBackendConnectionRead webBackendGetConnection(final WebBackendConnecti
*/
diff = refreshedCatalog.get().getCatalogDiff();
connection.setBreakingChange(refreshedCatalog.get().getBreakingChange());
connection.setStatus(refreshedCatalog.get().getConnectionStatus());
} else if (catalogUsedToMakeConfiguredCatalog.isPresent()) {
// reconstructs a full picture of the full schema at the time the catalog was configured.
syncCatalog = updateSchemaWithDiscovery(configuredCatalog, catalogUsedToMakeConfiguredCatalog.get());
Expand Down
Loading