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

Only migrate active and disable connection #9454

Merged
merged 3 commits into from
Jan 13, 2022
Merged

Only migrate active and disable connection #9454

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
90f011c
Only migrate active and disable connection
benmoriceau Jan 12, 2022
1fb68de
format
benmoriceau Jan 12, 2022
b0614f3
Merge branch 'master' of github.com:airbytehq/airbyte into bmoric/mig…
benmoriceau Jan 13, 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: 8 additions & 10 deletions airbyte-scheduler/app/src/main/java/io/airbyte/scheduler/app/SchedulerApp.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,13 +56,11 @@
import org.slf4j.MDC;

/**
* The SchedulerApp is responsible for finding new scheduled jobs that need to be run and to launch
* them. The current implementation uses two thread pools to do so. One pool is responsible for all
* job launching operations. The other pool is responsible for clean up operations.
* The SchedulerApp is responsible for finding new scheduled jobs that need to be run and to launch them. The current implementation uses two thread
* pools to do so. One pool is responsible for all job launching operations. The other pool is responsible for clean up operations.
* <p>
* Operations can have thread pools under the hood. An important thread pool to note is that the job
* submitter thread pool. This pool does the work of submitting jobs to temporal - the size of this
* pool determines the number of concurrent jobs that can be run. This is controlled via the
* Operations can have thread pools under the hood. An important thread pool to note is that the job submitter thread pool. This pool does the work of
* submitting jobs to temporal - the size of this pool determines the number of concurrent jobs that can be run. This is controlled via the
* SUBMITTER_NUM_THREADS variable of EnvConfigs.
*/
public class SchedulerApp {
Expand Down Expand Up @@ -140,7 +138,7 @@ public void start() throws IOException {
// anymore.
cleanupZombies(jobPersistence, jobNotifier);

LOGGER.error("Start running the old scheduler");
LOGGER.info("Start running the old scheduler");
scheduleJobsPool.scheduleWithFixedDelay(
() -> {
MDC.setContextMap(mdc);
Expand Down Expand Up @@ -234,13 +232,13 @@ public static void main(final String[] args) throws IOException, InterruptedExce
configs.getDatabaseUser(),
configs.getDatabasePassword(),
configs.getDatabaseUrl())
.getInitialized();
.getInitialized();
Copy link
Contributor

Choose a reason for hiding this comment

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

Probably need to run gw format

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Done


final Database configDatabase = new ConfigsDatabaseInstance(
configs.getConfigDatabaseUser(),
configs.getConfigDatabasePassword(),
configs.getConfigDatabaseUrl())
.getInitialized();
.getInitialized();
final ConfigPersistence configPersistence = new DatabaseConfigPersistence(configDatabase).withValidation();
final Optional<SecretPersistence> secretPersistence = SecretPersistence.getLongLived(configs);
final Optional<SecretPersistence> ephemeralSecretPersistence = SecretPersistence.getEphemeral(configs);
Expand Down Expand Up @@ -283,7 +281,7 @@ public static void main(final String[] args) throws IOException, InterruptedExce
Integer.parseInt(configs.getSubmitterNumThreads()),
configs.getSyncJobMaxAttempts(),
configs.getAirbyteVersionOrWarning(), configs.getWorkerEnvironment(), configs.getLogConfigs())
.start();
.start();
}

}
8 changes: 5 additions & 3 deletions 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 @@ -13,6 +13,7 @@
import io.airbyte.commons.version.AirbyteVersion;
import io.airbyte.config.Configs;
import io.airbyte.config.EnvConfigs;
import io.airbyte.config.StandardSync.Status;
import io.airbyte.config.helpers.LogClientSingleton;
import io.airbyte.config.init.YamlSeedConfigPersistence;
import io.airbyte.config.persistence.ConfigNotFoundException;
Expand Down Expand Up @@ -235,14 +236,15 @@ private static void migrateExistingConnection(final ConfigRepository configRepos
throws JsonValidationException, ConfigNotFoundException, IOException {
LOGGER.info("Start migration to the new scheduler...");
final Set<UUID> connectionIds =
configRepository.listStandardSyncs().stream().map(standardSync -> standardSync.getConnectionId()).collect(Collectors.toSet());
configRepository.listStandardSyncs().stream()
.filter(standardSync -> standardSync.getStatus() == Status.ACTIVE || standardSync.getStatus() == Status.INACTIVE)
Copy link
Contributor

Choose a reason for hiding this comment

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

What is the reasoning behind filtering down to only ACTIVE and INACTIVE?

Copy link
Contributor Author

@benmoriceau benmoriceau Jan 12, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

ACTIVE = enabled connections
INACTIVE = disabled connections
DEPRECATED = deleted connections

We need to start the enabled ones for an obvious reason. The disabled ones need to be started too so it can be re-enable later on. The workflow will await forever if it is disabled (like if it was a manual workflow).

Copy link
Contributor

Choose a reason for hiding this comment

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

I think that makes sense to me. Though we should probably get a confirmation from @jrhizor to be sure

.map(standardSync -> standardSync.getConnectionId()).collect(Collectors.toSet());
temporalWorkerRunFactory.migrateSyncIfNeeded(connectionIds);
LOGGER.info("Done migrating to the new scheduler...");
}

/**
* Copy paste from {@link io.airbyte.scheduler.app.SchedulerApp} which will be removed in a near
* future
* Copy paste from {@link io.airbyte.scheduler.app.SchedulerApp} which will be removed in a near future
*
* @param jobPersistence
* @param jobNotifier
Expand Down