Fixed empty previous run issue (perform fresh run when previous run not found or not-started for any reason). (#336)

Author: pravinbhat

RELEASE.md

@@ -1,4 +1,8 @@
 # Release Notes
+
+## [5.1.4] - 2024-12-04
+- Bug fix: Any run started with a `previousRunId` that is not found in the `cdm_run_info` table (for whatever reason), will be executed as a fresh new run instead of doing nothing.
+
 ## [5.1.3] - 2024-11-27
 - Bug fix: Fixed connection issue caused when using different types of origin and target clusters (e.g. Cassandra/DSE with host/port and Astra with SCB).
 

src/main/java/com/datastax/cdm/cql/statement/TargetUpsertRunDetailsStatement.java

@@ -101,11 +101,13 @@ public Collection<PartitionRange> getPendingPartitions(long prevRunId, JobType j
                 .execute(boundSelectInfoStatement.setString("table_name", tableName).setLong("run_id", prevRunId));
         Row cdmRunStatus = rsInfo.one();
         if (cdmRunStatus == null) {
-            return Collections.emptyList();
+            throw new RunNotStartedException(
+                    "###################### Run NOT FOUND for Previous RunId: " + prevRunId + ", starting new run!");
         } else {
             String status = cdmRunStatus.getString("status");
             if (TrackRun.RUN_STATUS.NOT_STARTED.toString().equals(status)) {
-                throw new RunNotStartedException("Run not started for run_id: " + prevRunId);
+                throw new RunNotStartedException("###################### Run NOT STARTED for Previous RunId: "
+                        + prevRunId + ", starting new run!");
             }
         }
 

src/main/scala/com/datastax/cdm/job/BaseJob.scala

@@ -33,7 +33,7 @@ import scala.collection.JavaConverters._
 
 abstract class BaseJob[T: ClassTag] extends App {
 
-  private val abstractLogger = LoggerFactory.getLogger(this.getClass.getName)
+  protected val abstractLogger = LoggerFactory.getLogger(this.getClass.getName)
 
   private var jobName: String = _
   var jobFactory: IJobSessionFactory[T] = _

src/main/scala/com/datastax/cdm/job/BasePartitionJob.scala

@@ -42,7 +42,10 @@ abstract class BasePartitionJob extends BaseJob[PartitionRange] {
       try {
         trackRunFeature.getPendingPartitions(prevRunId, jobType)
       } catch {
-        case e: RunNotStartedException => SplitPartitions.getRandomSubPartitions(pieces, minPartition, maxPartition, coveragePercent, jobType)
+        case e: RunNotStartedException => {
+          abstractLogger.warn(e.getMessage)
+          SplitPartitions.getRandomSubPartitions(pieces, minPartition, maxPartition, coveragePercent, jobType)
+        }
       }
     } else {
       SplitPartitions.getRandomSubPartitions(pieces, minPartition, maxPartition, coveragePercent, jobType)

src/test/java/com/datastax/cdm/cql/statement/TargetUpsertRunDetailsStatementTest.java

@@ -32,6 +32,7 @@
 import org.mockito.Mock;
 
 import com.datastax.cdm.cql.CommonMocks;
+import com.datastax.cdm.feature.TrackRun;
 import com.datastax.cdm.job.IJobSessionFactory.JobType;
 import com.datastax.cdm.job.PartitionRange;
 import com.datastax.cdm.job.RunNotStartedException;
@@ -72,15 +73,43 @@ public void setup() {
     }
 
     @Test
-    public void getPendingPartitions_nothingPending() throws RunNotStartedException {
+    public void incorrectKsTable() {
+        assertThrows(RuntimeException.class, () -> new TargetUpsertRunDetailsStatement(cqlSession, "table1"));
+    }
+
+    @Test
+    public void getPendingPartitions_noPrevRun() throws RunNotStartedException {
         targetUpsertRunDetailsStatement = new TargetUpsertRunDetailsStatement(cqlSession, "ks.table1");
         assertEquals(Collections.emptyList(), targetUpsertRunDetailsStatement.getPendingPartitions(0, JobType.MIGRATE));
-        assertEquals(Collections.emptyList(), targetUpsertRunDetailsStatement.getPendingPartitions(1, JobType.MIGRATE));
     }
 
     @Test
-    public void incorrectKsTable() throws RunNotStartedException {
-        assertThrows(RuntimeException.class, () -> new TargetUpsertRunDetailsStatement(cqlSession, "table1"));
+    public void getPendingPartitions_noPrevRunFound() {
+        targetUpsertRunDetailsStatement = new TargetUpsertRunDetailsStatement(cqlSession, "ks.table1");
+        assertThrows(RunNotStartedException.class,
+                () -> targetUpsertRunDetailsStatement.getPendingPartitions(1, JobType.MIGRATE));
+    }
+
+    @Test
+    public void getPendingPartitions_prevRunNotStarted() {
+        when(rs.one()).thenReturn(row1);
+        when(row1.getString("status")).thenReturn(TrackRun.RUN_STATUS.NOT_STARTED.toString());
+
+        targetUpsertRunDetailsStatement = new TargetUpsertRunDetailsStatement(cqlSession, "ks.table1");
+        assertThrows(RunNotStartedException.class,
+                () -> targetUpsertRunDetailsStatement.getPendingPartitions(123, JobType.MIGRATE));
+    }
+
+    @Test
+    public void getPendingPartitions_prevRunNoPartsPending() throws RunNotStartedException {
+        when(rs.one()).thenReturn(row1);
+        when(row1.getString("status")).thenReturn(TrackRun.RUN_STATUS.ENDED.toString());
+        Iterator mockIterator = mock(Iterator.class);
+        when(rs.iterator()).thenReturn(mockIterator);
+        when(mockIterator.hasNext()).thenReturn(false);
+
+        targetUpsertRunDetailsStatement = new TargetUpsertRunDetailsStatement(cqlSession, "ks.table1");
+        assertEquals(Collections.emptyList(), targetUpsertRunDetailsStatement.getPendingPartitions(123, JobType.MIGRATE));
     }
 
     @Test