Revert "Fix Worker Request flow to properly use batching (#775)" (#785)

This reverts commit 3b0c92f.

Author: Andyz26

mantis-control-plane/mantis-control-plane-server/src/main/java/io/mantisrx/master/jobcluster/job/JobActor.java

@@ -2502,7 +2502,6 @@ public int scaleStage(
                             getJobId(), getJobState()));
                     throw new RuntimeException(error);
                 }
-                List<IMantisWorkerMetadata> workerRequests = new ArrayList<>();
                 if (newNumWorkerCount > oldNumWorkers) {
                     for (int i = 0; i < newNumWorkerCount - oldNumWorkers; i++) {
                         try {
@@ -2511,17 +2510,14 @@ public int scaleStage(
                             IMantisWorkerMetadata workerRequest = addWorker(schedInfo, stageMetaData.getStageNum(),
                                     newWorkerIndex);
                             jobStore.storeNewWorker(workerRequest);
-                            markStageAssignmentsChanged(true);
-                            workerRequests.add(workerRequest);
+                            queueTask(workerRequest);
                         } catch (Exception e) {
                             // creating a worker failed but expected no of workers was set successfully,
                             // during heartbeat check we will
                             // retry launching this worker
                             LOGGER.warn("Exception adding new worker for {}", stageMetaData.getJobId().getId(), e);
                         }
                     }
-                    //one request to provision all new workers
-                    queueTasks(workerRequests, empty());
                 } else {
                     //  potential bulk removal opportunity?
                     for (int i = 0; i < oldNumWorkers - newNumWorkerCount; i++) {

mantis-control-plane/mantis-control-plane-server/src/main/java/io/mantisrx/master/resourcecluster/ExecutorStateManagerImpl.java

@@ -502,7 +502,6 @@ public GetClusterUsageResponse getClusterUsage(GetClusterUsageRequest req) {
             }
         });
 
-        log.debug("Pending TE count by group key: {}" ,pendingCountByGroupKey);
         // remove jobs from pending set which have all pending workers
         jobIdToMachineDef.forEach((jobId, workers) -> {
             final JobRequirements jobStats = pendingJobRequests.getIfPresent(jobId);
@@ -567,9 +566,6 @@ private Optional<Map<TaskExecutorID, TaskExecutorState>> findTaskExecutorsFor(Ta
                     pendingJobRequests.put(request.getJobId(), new JobRequirements(request.getGroupedBySchedulingConstraints()));
                 }
             }
-            else {
-                log.info("Not adding job {} to pending requests for {} scheduling constraints {} because there were less than 2 TE allocation requests", request.getJobId(), allocationRequests.size(), schedulingConstraints);
-            }
             return Optional.empty();
         }
     }

mantis-control-plane/mantis-control-plane-server/src/test/java/io/mantisrx/master/jobcluster/job/JobScaleUpDownTests.java

@@ -145,8 +145,7 @@ public void testJobScaleUp() throws Exception, InvalidJobException, io.mantisrx.
 		verify(jobStoreMock, times(3)).updateJob(any());
 
 		// initial worker + job master and scale up worker
-        //should be twice because it is an initial request + scale up request
-		verify(schedulerMock, times(2)).scheduleWorkers(any());
+		verify(schedulerMock, times(3)).scheduleWorkers(any());
 
 	}