Randomized the pending token-range list (#354)

* Randomized the pending token-range list returned by the `trackRun` feature
* Update src/main/java/com/datastax/cdm/cql/statement/TargetUpsertRunDetailsStatement.java

Agree, this looks cleaner

Co-authored-by: Madhavan <[email protected]>

* Apply suggestions from code review

---------

Co-authored-by: Madhavan <[email protected]>

Author: pravinbhat

RELEASE.md

@@ -1,5 +1,8 @@
 # Release Notes
 
+## [5.2.3] - 2025-04-15
+- Randomized the pending token-range list returned by the `trackRun` feature (when rerunning a previously incomplete job) for better load distribution across the cluster.
+
 ## [5.2.2] - 2025-04-02
 - Replaced deprecated (not supported in Cassandra 5.0.3+) function `dateof()` with `totimestamp()`
 

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

@@ -20,6 +20,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -111,11 +112,19 @@ public Collection<PartitionRange> getPendingPartitions(long prevRunId, JobType j
             }
         }
 
-        final Collection<PartitionRange> pendingParts = new ArrayList<PartitionRange>();
-        pendingParts.addAll(getPartitionsByStatus(prevRunId, TrackRun.RUN_STATUS.NOT_STARTED.toString(), jobType));
-        pendingParts.addAll(getPartitionsByStatus(prevRunId, TrackRun.RUN_STATUS.STARTED.toString(), jobType));
-        pendingParts.addAll(getPartitionsByStatus(prevRunId, TrackRun.RUN_STATUS.FAIL.toString(), jobType));
-        pendingParts.addAll(getPartitionsByStatus(prevRunId, TrackRun.RUN_STATUS.DIFF.toString(), jobType));
+        final List<PartitionRange> pendingParts = new ArrayList<PartitionRange>();
+        // Use an array of statuses for iteration
+        String[] statuses = {
+            TrackRun.RUN_STATUS.NOT_STARTED.toString(),
+            TrackRun.RUN_STATUS.STARTED.toString(),
+            TrackRun.RUN_STATUS.FAIL.toString(),
+            TrackRun.RUN_STATUS.DIFF.toString()
+        };
+        for (String status : statuses) {
+            pendingParts.addAll(getPartitionsByStatus(prevRunId, status, jobType));
+        }
+        Collections.shuffle(pendingParts);
+        Collections.shuffle(pendingParts);
 
         return pendingParts;
     }

src/main/java/com/datastax/cdm/job/SplitPartitions.java

@@ -35,8 +35,6 @@ public static List<PartitionRange> getRandomSubPartitions(int numSplits, BigInte
         List<PartitionRange> partitions = getSubPartitions(numSplits, min, max, coveragePercent, jobType);
         Collections.shuffle(partitions);
         Collections.shuffle(partitions);
-        Collections.shuffle(partitions);
-        Collections.shuffle(partitions);
         return partitions;
     }