Remove unused job persistence methods. (#18952)

Closes #12823.

Also remove unused dump methods.

Author: davinchia

airbyte-persistence/job-persistence/src/main/java/io/airbyte/persistence/job/DefaultJobPersistence.java

@@ -91,9 +91,6 @@ public class DefaultJobPersistence implements JobPersistence {
   private final int JOB_HISTORY_EXCESSIVE_NUMBER_OF_JOBS;
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DefaultJobPersistence.class);
-  private static final Set<String> SYSTEM_SCHEMA = Set
-      .of("pg_toast", "information_schema", "pg_catalog", "import_backup", "pg_internal",
-          "catalog_history");
   public static final String ATTEMPT_NUMBER = "attempt_number";
   private static final String JOB_ID = "job_id";
   private static final String WHERE = "WHERE ";
@@ -797,18 +794,6 @@ public void setSecretMigrationDone() throws IOException {
     setMetadata(SECRET_MIGRATION_STATUS, "true");
   }
 
-  private final String SCHEDULER_MIGRATION_STATUS = "schedulerMigration";
-
-  @Override
-  public boolean isSchedulerMigrated() throws IOException {
-    return getMetadata(SCHEDULER_MIGRATION_STATUS).count() == 1;
-  }
-
-  @Override
-  public void setSchedulerMigrationDone() throws IOException {
-    setMetadata(SCHEDULER_MIGRATION_STATUS, "true");
-  }
-
   @Override
   public Optional<String> getVersion() throws IOException {
     return getMetadata(AirbyteVersion.AIRBYTE_VERSION_KEY_NAME).findFirst();
@@ -915,27 +900,6 @@ public Map<JobsDatabaseSchema, Stream<JsonNode>> exportDatabase() throws IOExcep
     return exportDatabase(DEFAULT_SCHEMA);
   }
 
-  /**
-   * This is different from {@link #exportDatabase()} cause it exports all the tables in all the
-   * schemas available
-   */
-  @Override
-  public Map<String, Stream<JsonNode>> dump() throws IOException {
-    final Map<String, Stream<JsonNode>> result = new HashMap<>();
-    for (final String schema : listSchemas()) {
-      final List<String> tables = listAllTables(schema);
-
-      for (final String table : tables) {
-        if (result.containsKey(table)) {
-          throw new RuntimeException("Multiple tables found with the same name " + table);
-        }
-        result.put(table.toUpperCase(), exportTable(schema, table));
-      }
-    }
-
-    return result;
-  }
-
   private Map<JobsDatabaseSchema, Stream<JsonNode>> exportDatabase(final String schema) throws IOException {
     final List<String> tables = listTables(schema);
     final Map<JobsDatabaseSchema, Stream<JsonNode>> result = new HashMap<>();
@@ -982,25 +946,6 @@ public void purgeJobHistory(final LocalDateTime asOfDate) {
     }
   }
 
-  private List<String> listAllTables(final String schema) throws IOException {
-    if (schema != null) {
-      return jobDatabase.query(context -> context.meta().getSchemas(schema).stream()
-          .flatMap(s -> context.meta(s).getTables().stream())
-          .map(Named::getName)
-          .collect(Collectors.toList()));
-    } else {
-      return List.of();
-    }
-  }
-
-  private List<String> listSchemas() throws IOException {
-    return jobDatabase.query(context -> context.meta().getSchemas().stream()
-        .map(Named::getName)
-        .filter(c -> !SYSTEM_SCHEMA.contains(c))
-        .collect(Collectors.toList()));
-
-  }
-
   private Stream<JsonNode> exportTable(final String schema, final String tableName) throws IOException {
     final Table<Record> tableSql = getTable(schema, tableName);
     try (final Stream<Record> records = jobDatabase.query(ctx -> ctx.select(DSL.asterisk()).from(tableSql).fetchStream())) {

airbyte-persistence/job-persistence/src/main/java/io/airbyte/persistence/job/JobPersistence.java

@@ -280,8 +280,6 @@ List<JobWithStatusAndTimestamp> listJobStatusAndTimestampWithConnection(UUID con
    */
   Map<JobsDatabaseSchema, Stream<JsonNode>> exportDatabase() throws IOException;
 
-  Map<String, Stream<JsonNode>> dump() throws IOException;
-
   /**
    * Import all SQL tables from streams of JsonNode objects.
    *
@@ -306,22 +304,6 @@ List<JobWithStatusAndTimestamp> listJobStatusAndTimestampWithConnection(UUID con
    */
   void setSecretMigrationDone() throws IOException;
 
-  /**
-   * Check if the scheduler has been migrated to temporal.
-   *
-   * TODO (https://github.com/airbytehq/airbyte/issues/12823): remove this method after the next
-   * "major" version bump as it will no longer be needed.
-   */
-  boolean isSchedulerMigrated() throws IOException;
-
-  /**
-   * Set that the scheduler migration has been performed.
-   *
-   * TODO (https://github.com/airbytehq/airbyte/issues/12823): remove this method after the next
-   * "major" version bump as it will no longer be needed.
-   */
-  void setSchedulerMigrationDone() throws IOException;
-
   List<AttemptNormalizationStatus> getAttemptNormalizationStatusesForJob(final Long jobId) throws IOException;
 
 }

airbyte-persistence/job-persistence/src/test/java/io/airbyte/persistence/job/DefaultJobPersistenceTest.java

@@ -566,15 +566,6 @@ void testSecretMigrationMetadata() throws IOException {
     assertTrue(isMigrated);
   }
 
-  @Test
-  void testSchedulerMigrationMetadata() throws IOException {
-    boolean isMigrated = jobPersistence.isSchedulerMigrated();
-    assertFalse(isMigrated);
-    jobPersistence.setSchedulerMigrationDone();
-    isMigrated = jobPersistence.isSchedulerMigrated();
-    assertTrue(isMigrated);
-  }
-
   @Test
   void testAirbyteProtocolVersionMaxMetadata() throws IOException {
     assertTrue(jobPersistence.getAirbyteProtocolVersionMax().isEmpty());

airbyte-server/src/main/java/io/airbyte/server/ServerApp.java

@@ -4,7 +4,6 @@
 
 package io.airbyte.server;
 
-import com.google.common.annotations.VisibleForTesting;
 import io.airbyte.analytics.Deployment;
 import io.airbyte.analytics.TrackingClient;
 import io.airbyte.analytics.TrackingClientSingleton;
@@ -18,10 +17,7 @@
 import io.airbyte.commons.version.AirbyteVersion;
 import io.airbyte.config.Configs;
 import io.airbyte.config.EnvConfigs;
-import io.airbyte.config.StandardSync;
-import io.airbyte.config.StandardSync.Status;
 import io.airbyte.config.helpers.LogClientSingleton;
-import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.config.persistence.SecretsRepositoryReader;
 import io.airbyte.config.persistence.SecretsRepositoryWriter;
@@ -70,16 +66,12 @@
 import io.airbyte.server.scheduler.EventRunner;
 import io.airbyte.server.scheduler.TemporalEventRunner;
 import io.airbyte.validation.json.JsonSchemaValidator;
-import io.airbyte.validation.json.JsonValidationException;
 import io.airbyte.workers.normalization.NormalizationRunnerFactory;
 import io.temporal.serviceclient.WorkflowServiceStubs;
-import java.io.IOException;
 import java.net.http.HttpClient;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.UUID;
-import java.util.stream.Collectors;
 import javax.sql.DataSource;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.servlet.ServletContextHandler;
@@ -256,14 +248,6 @@ public static ServerRunnable getServer(final ServerFactory apiFactory,
     final HttpClient httpClient = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build();
     final EventRunner eventRunner = new TemporalEventRunner(temporalClient);
 
-    // It is important that the migration to the temporal scheduler is performed before the server
-    // accepts any requests.
-    // This is why this migration is performed here instead of in the bootloader - so that the server
-    // blocks on this.
-    // TODO (https://github.com/airbytehq/airbyte/issues/12823): remove this method after the next
-    // "major" version bump as it will no longer be needed.
-    migrateExistingConnectionsToTemporalScheduler(configRepository, jobPersistence, eventRunner);
-
     final WorkspaceHelper workspaceHelper = new WorkspaceHelper(configRepository, jobPersistence);
 
     final JsonSchemaValidator schemaValidator = new JsonSchemaValidator();
@@ -370,28 +354,6 @@ public static ServerRunnable getServer(final ServerFactory apiFactory,
         workspacesHandler);
   }
 
-  @VisibleForTesting
-  static void migrateExistingConnectionsToTemporalScheduler(final ConfigRepository configRepository,
-                                                            final JobPersistence jobPersistence,
-                                                            final EventRunner eventRunner)
-      throws JsonValidationException, ConfigNotFoundException, IOException {
-    // Skip the migration if it was already performed, to save on resources/startup time
-    if (jobPersistence.isSchedulerMigrated()) {
-      LOGGER.info("Migration to temporal scheduler has already been performed");
-      return;
-    }
-
-    LOGGER.info("Start migration to the new scheduler...");
-    final Set<UUID> connectionIds =
-        configRepository.listStandardSyncs().stream()
-            .filter(standardSync -> standardSync.getStatus() == Status.ACTIVE || standardSync.getStatus() == Status.INACTIVE)
-            .map(StandardSync::getConnectionId)
-            .collect(Collectors.toSet());
-    eventRunner.migrateSyncIfNeeded(connectionIds);
-    jobPersistence.setSchedulerMigrationDone();
-    LOGGER.info("Done migrating to the new scheduler...");
-  }
-
   public static void main(final String[] args) {
     try {
       final Configs configs = new EnvConfigs();