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

Jump to bottom

Check disabled connections after protocol update #18990

Merged
merged 5 commits into from
Nov 8, 2022
Merged

Check disabled connections after protocol update #18990

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ed6add0
Add support to clear unsupported protocol flag to StandardSyncPersist…
gosusnp Nov 4, 2022
0be3711
Fix internal records scope
gosusnp Nov 4, 2022
f2305ea
Plug clearUnsupportedProtocolVersionFlag on connector update
gosusnp Nov 4, 2022
535878f
Update UPDATED_AT when clearing protocol flags
gosusnp Nov 4, 2022
5e472c9
Merge branch 'master' into gosusnp/16043-check-disabled-connections-a…
gosusnp Nov 7, 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
18 changes: 18 additions & 0 deletions ...nfig/config-persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import io.airbyte.commons.json.Jsons;
import io.airbyte.commons.lang.MoreBooleans;
import io.airbyte.commons.version.AirbyteProtocolVersion;
import io.airbyte.commons.version.AirbyteProtocolVersionRange;
import io.airbyte.config.ActorCatalog;
import io.airbyte.config.ActorCatalogFetchEvent;
import io.airbyte.config.ConfigSchema;
Expand Down Expand Up @@ -728,6 +729,23 @@ public void writeStandardSync(final StandardSync standardSync) throws JsonValida
standardSyncPersistence.writeStandardSync(standardSync);
}

/**
* For the StandardSyncs related to actorDefinitionId, clear the unsupported protocol version flag
* if both connectors are now within support range.
*
* @param actorDefinitionId the actorDefinitionId to query
* @param actorType the ActorType of actorDefinitionId
* @param supportedRange the supported range of protocol versions
*/
// We have conflicting imports here, ActorType is imported from jooq for most internal uses. Since
// this is a public method, we should be using the ActorType from airbyte-config.
public void clearUnsupportedProtocolVersionFlag(final UUID actorDefinitionId,
final io.airbyte.config.ActorType actorType,
final AirbyteProtocolVersionRange supportedRange)
throws IOException {
standardSyncPersistence.clearUnsupportedProtocolVersionFlag(actorDefinitionId, actorType, supportedRange);
}

public List<StandardSync> listStandardSyncs() throws IOException, JsonValidationException {
return standardSyncPersistence.listStandardSync();
}
Expand Down
76 changes: 76 additions & 0 deletions ...nfig-persistence/src/main/java/io/airbyte/config/persistence/StandardSyncPersistence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

package io.airbyte.config.persistence;

import static io.airbyte.db.instance.configs.jooq.generated.Tables.ACTOR;
import static io.airbyte.db.instance.configs.jooq.generated.Tables.ACTOR_DEFINITION;
import static io.airbyte.db.instance.configs.jooq.generated.Tables.CONNECTION;
import static io.airbyte.db.instance.configs.jooq.generated.Tables.CONNECTION_OPERATION;
import static io.airbyte.db.instance.configs.jooq.generated.Tables.STATE;
Expand All @@ -12,18 +14,25 @@

import io.airbyte.commons.enums.Enums;
import io.airbyte.commons.json.Jsons;
import io.airbyte.commons.version.AirbyteProtocolVersion;
import io.airbyte.commons.version.AirbyteProtocolVersionRange;
import io.airbyte.commons.version.Version;
import io.airbyte.config.ActorType;
import io.airbyte.config.ConfigSchema;
import io.airbyte.config.ConfigWithMetadata;
import io.airbyte.config.StandardSync;
import io.airbyte.config.helpers.ScheduleHelpers;
import io.airbyte.db.Database;
import io.airbyte.db.ExceptionWrappingDatabase;
import io.airbyte.db.instance.configs.jooq.generated.tables.Actor;
import io.airbyte.db.instance.configs.jooq.generated.tables.ActorDefinition;
import java.io.IOException;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.stream.Stream;
import org.jooq.DSLContext;
import org.jooq.JSONB;
import org.jooq.Record;
Expand All @@ -32,6 +41,13 @@

public class StandardSyncPersistence {

private record StandardSyncIdsWithProtocolVersions(
UUID standardSyncId,
UUID sourceDefId,
Version sourceProtocolVersion,
UUID destinationDefId,
Version destinationProtocolVersion) {}

private final ExceptionWrappingDatabase database;

public StandardSyncPersistence(final Database database) {
Expand Down Expand Up @@ -78,6 +94,29 @@ public void deleteStandardSync(final UUID standardSyncId) throws IOException {
});
}

/**
* For the StandardSyncs related to actorDefinitionId, clear the unsupported protocol version flag
* if both connectors are now within support range.
*
* @param actorDefinitionId the actorDefinitionId to query
* @param actorType the ActorType of actorDefinitionId
* @param supportedRange the supported range of protocol versions
*/
public void clearUnsupportedProtocolVersionFlag(final UUID actorDefinitionId,
final ActorType actorType,
final AirbyteProtocolVersionRange supportedRange)
throws IOException {
final Stream<StandardSyncIdsWithProtocolVersions> candidateSyncs = database.query(ctx -> findDisabledSyncs(ctx, actorDefinitionId, actorType));
final List<UUID> standardSyncsToReEnable = candidateSyncs
.filter(sync -> supportedRange.isSupported(sync.sourceProtocolVersion()) && supportedRange.isSupported(sync.destinationProtocolVersion()))
.map(StandardSyncIdsWithProtocolVersions::standardSyncId)
.toList();
database.query(ctx -> {
clearProtocolVersionFlag(ctx, standardSyncsToReEnable);
return null;
});
}

private List<ConfigWithMetadata<StandardSync>> listStandardSyncWithMetadata(final Optional<UUID> configId) throws IOException {
final Result<Record> result = database.query(ctx -> {
final SelectJoinStep<Record> query = ctx.select(asterisk()).from(CONNECTION);
Expand Down Expand Up @@ -214,4 +253,41 @@ private void writeStandardSync(final StandardSync standardSync, final DSLContext
}
}

private Stream<StandardSyncIdsWithProtocolVersions> findDisabledSyncs(final DSLContext ctx, final UUID actorDefId, final ActorType actorType) {
// Table aliasing to help have a readable join
final Actor source = ACTOR.as("source");
final Actor destination = ACTOR.as("destination");
final ActorDefinition sourceDef = ACTOR_DEFINITION.as("sourceDef");
final ActorDefinition destDef = ACTOR_DEFINITION.as("destDef");

// Retrieve all the connections currently disabled due to a bad protocol version
// where the actor definition is matching the one provided to this function
final Stream<StandardSyncIdsWithProtocolVersions> results = ctx
.select(CONNECTION.ID, sourceDef.ID, sourceDef.PROTOCOL_VERSION, destDef.ID, destDef.PROTOCOL_VERSION)
.from(CONNECTION)
.join(source).on(CONNECTION.SOURCE_ID.eq(source.ID))
.join(sourceDef).on(source.ACTOR_DEFINITION_ID.eq(sourceDef.ID))
Copy link
Contributor

Choose a reason for hiding this comment

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

Why do we need the join on the source / source def ? I think that it is missing in the where clause.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

To keep things simple, my approach here was to always retrieve (ConnectionId, sourceDefId, sourceProtocolVersion, destDefId, destProtocolVersion).
Since protocol version is store in the actor definition, we always need to join both sides (source+destination).

An alternative could be to have some specific logic to only filter source or destination, however, I think it adds a few more conditional blocks overall throughout the code base for an operation that in the end should be relatively rare.

.join(destination).on(CONNECTION.DESTINATION_ID.eq(destination.ID))
.join(destDef).on(destination.ACTOR_DEFINITION_ID.eq(destDef.ID))
.where(
CONNECTION.UNSUPPORTED_PROTOCOL_VERSION.eq(true).and(
(actorType == ActorType.DESTINATION ? destDef : sourceDef).ID.eq(actorDefId)))
Copy link
Contributor

Choose a reason for hiding this comment

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

Should we do the same for the Source Actor type?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

This is what the inline if is doing here, based on the actorType it will either add where source.ID.eq(actorDefId) or destination.ID.eq(actorDefId)

.fetchStream()
.map(r -> new StandardSyncIdsWithProtocolVersions(
r.get(CONNECTION.ID),
r.get(sourceDef.ID),
AirbyteProtocolVersion.getWithDefault(r.get(sourceDef.PROTOCOL_VERSION)),
r.get(destDef.ID),
AirbyteProtocolVersion.getWithDefault(r.get(destDef.PROTOCOL_VERSION))));
return results;
}

private void clearProtocolVersionFlag(final DSLContext ctx, final List<UUID> standardSyncIds) {
ctx.update(CONNECTION)
Copy link
Contributor

Choose a reason for hiding this comment

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

Nit: Please update the modified_at collumn as well. I don't think that is being updated through a trigger.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Good catch, updated.

.set(CONNECTION.UNSUPPORTED_PROTOCOL_VERSION, false)
.set(CONNECTION.UPDATED_AT, OffsetDateTime.now())
.where(CONNECTION.ID.in(standardSyncIds))
.execute();
}

}
Loading