refactor: remove unused methods in JobPersistence (#16958)

Author: cgardens

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

@@ -638,17 +638,6 @@ public Optional<Long> enqueueJob(final String scope, final JobConfig jobConfig,
         .map(r -> r.getValue("id", Long.class));
   }
 
-  // TODO: This is unused outside of test. Need to remove it.
-  @Override
-  public void resetJob(final long jobId) throws IOException {
-    // TODO: stop using LocalDateTime
-    // https://github.com/airbytehq/airbyte-platform-internal/issues/10815
-    jobDatabase.query(ctx -> {
-      updateJobStatus(ctx, jobId, JobStatus.PENDING);
-      return null;
-    });
-  }
-
   @Override
   public void cancelJob(final long jobId) throws IOException {
     // TODO: stop using LocalDateTime
@@ -960,7 +949,7 @@ public Long getJobCount(final Set<ConfigType> configTypes,
         .fetchOne().into(Long.class));
   }
 
-  public Result<Record> listJobsQuery(final Set<ConfigType> configTypes, final String configId, final int pagesize, String orderByString)
+  private Result<Record> listJobsQuery(final Set<ConfigType> configTypes, final String configId, final int pagesize, String orderByString)
       throws IOException {
     return jobDatabase.query(ctx -> {
       final String jobsSubquery = "(" + ctx.select(DSL.asterisk()).from(JOBS)
@@ -1395,19 +1384,6 @@ public Optional<Job> getFirstReplicationJob(final UUID connectionId) throws IOEx
         .flatMap(r -> getJobOptional(ctx, r.get("id", Long.class))));
   }
 
-  @VisibleForTesting
-  List<AttemptWithJobInfo> listAttemptsWithJobInfo(final ConfigType configType, final Instant attemptEndedAtTimestamp, final int limit)
-      throws IOException {
-    // TODO: stop using LocalDateTime
-    // https://github.com/airbytehq/airbyte-platform-internal/issues/10815
-    final LocalDateTime timeConvertedIntoLocalDateTime = convertInstantToLocalDataTime(attemptEndedAtTimestamp);
-    return jobDatabase.query(ctx -> getAttemptsWithJobsFromResult(ctx.fetch(
-        BASE_JOB_SELECT_AND_JOIN + WHERE + "CAST(config_type AS VARCHAR) =  ? AND " + " attempts.ended_at > ? ORDER BY attempts.ended_at ASC LIMIT ?",
-        toSqlName(configType),
-        timeConvertedIntoLocalDateTime,
-        limit)));
-  }
-
   @Override
   public Optional<String> getVersion() throws IOException {
     return getMetadata(AirbyteVersion.AIRBYTE_VERSION_KEY_NAME).findFirst();

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

@@ -89,15 +89,6 @@ public interface JobPersistence {
    */
   Optional<Long> enqueueJob(String scope, JobConfig jobConfig, boolean isScheduled) throws IOException;
 
-  /**
-   * Set job status from current status to PENDING. Throws {@link IllegalStateException} if the job is
-   * in a terminal state.
-   *
-   * @param jobId job to reset
-   * @throws IOException exception due to interaction with persistence
-   */
-  void resetJob(long jobId) throws IOException;
-
   //
   // JOB LIFECYCLE
   //

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

@@ -1671,9 +1671,8 @@ void testGetFirstSyncJobForConnectionIdEmpty() throws IOException {
     void testGetFirstSyncJobForConnectionId() throws IOException {
       final long jobId1 = jobPersistence.enqueueJob(SCOPE, SYNC_JOB_CONFIG, true).orElseThrow();
       jobPersistence.succeedAttempt(jobId1, jobPersistence.createAttempt(jobId1, LOG_PATH));
-      final List<AttemptWithJobInfo> attemptsWithJobInfo =
-          jobPersistence.listAttemptsWithJobInfo(SYNC_JOB_CONFIG.getConfigType(), Instant.EPOCH, 1000);
-      final List<Attempt> attempts = Collections.singletonList(attemptsWithJobInfo.get(0).attempt);
+      final List<Attempt> attemptsWithJobInfo = jobPersistence.getJob(jobId1).attempts;
+      final List<Attempt> attempts = Collections.singletonList(attemptsWithJobInfo.get(0));
 
       final Instant afterNow = NOW.plusSeconds(1000);
       when(timeSupplier.get()).thenReturn(afterNow);
@@ -2261,38 +2260,4 @@ void testListJobsWithStatusesAndConfigTypesForConnection() throws IOException, I
 
   }
 
-  @Nested
-  @DisplayName("When resetting job")
-  class ResetJob {
-
-    @Test
-    @DisplayName("Should reset job and put job in pending state")
-    void testResetJob() throws IOException {
-      final long jobId = jobPersistence.enqueueJob(SCOPE, SPEC_JOB_CONFIG, true).orElseThrow();
-      final int attemptNumber = jobPersistence.createAttempt(jobId, LOG_PATH);
-      final Job created = jobPersistence.getJob(jobId);
-
-      jobPersistence.failAttempt(jobId, attemptNumber);
-      when(timeSupplier.get()).thenReturn(Instant.ofEpochMilli(4242));
-      jobPersistence.resetJob(jobId);
-
-      final Job updated = jobPersistence.getJob(jobId);
-      assertEquals(JobStatus.PENDING, updated.status);
-      assertNotEquals(created.updatedAtInSecond, updated.updatedAtInSecond);
-    }
-
-    @Test
-    @DisplayName("Should not be able to reset a cancelled job")
-    void testResetJobCancelled() throws IOException {
-      final long jobId = jobPersistence.enqueueJob(SCOPE, SPEC_JOB_CONFIG, true).orElseThrow();
-
-      jobPersistence.cancelJob(jobId);
-      assertDoesNotThrow(() -> jobPersistence.resetJob(jobId));
-
-      final Job updated = jobPersistence.getJob(jobId);
-      assertEquals(JobStatus.CANCELLED, updated.status);
-    }
-
-  }
-
 }