Skip to content

Remove deprecated job type ctor #837

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Andyz26 merged 1 commit into from
Mar 24, 2026
Merged

Remove deprecated job type ctor #837

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 0 additions & 12 deletions ...rc/main/java/io/mantisrx/server/master/resourcecluster/TaskExecutorAllocationRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,16 +42,4 @@ public static TaskExecutorAllocationRequest of(
return new TaskExecutorAllocationRequest(workerId, constraints, jobMetadata, stageNum, -1L, durationType);
}

/**
* @deprecated Use {@link #of(WorkerId, SchedulingConstraints, JobMetadata, int, MantisJobDurationType)} instead.
* This overload hardcodes durationType to Perpetual, which causes subscription timeout to not fire for Transient jobs.
*/
@Deprecated
public static TaskExecutorAllocationRequest of(
WorkerId workerId,
SchedulingConstraints constraints,
JobMetadata jobMetadata,
int stageNum) {
return new TaskExecutorAllocationRequest(workerId, constraints, jobMetadata, stageNum, -1L, MantisJobDurationType.Perpetual);
}
}
3 changes: 2 additions & 1 deletion ...mantis-control-plane-server/src/main/java/io/mantisrx/master/jobcluster/job/JobActor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1939,7 +1939,8 @@ void upsertReservationForStage(
sr.getWorkerId(),
sr.getSchedulingConstraints(),
sr.getJobMetadata(),
sr.getStageNum()))
sr.getStageNum(),
sr.getDurationType()))
.collect(Collectors.toSet());

// Build reservation request using shared UpsertReservation class
Expand Down
75 changes: 38 additions & 37 deletions ...ne-server/src/test/java/io/mantisrx/master/resourcecluster/ExecutorStateManagerTests.java
View file
Open in desktop

Large diffs are not rendered by default.

15 changes: 8 additions & 7 deletions ...test/java/io/mantisrx/master/resourcecluster/ReservationRegistryActorIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import static io.mantisrx.server.master.resourcecluster.proto.MantisResourceClusterReservationProto.*;
import io.mantisrx.master.scheduler.CpuWeightedFitnessCalculator;
import io.mantisrx.runtime.MachineDefinition;
import io.mantisrx.runtime.MantisJobDurationType;
import io.mantisrx.server.core.TestingRpcService;
import io.mantisrx.server.core.domain.WorkerId;
import io.mantisrx.server.core.scheduler.SchedulingConstraints;
Expand Down Expand Up @@ -201,8 +202,8 @@ public void testFullReservationWorkflowWithTaskExecutorAllocation() throws Excep
WorkerId worker1 = WorkerId.fromIdUnsafe("job-1-worker-0-1");
WorkerId worker2 = WorkerId.fromIdUnsafe("job-2-worker-0-1");

TaskExecutorAllocationRequest req1 = TaskExecutorAllocationRequest.of(worker1, constraints, createJobMetadata(), 1);
TaskExecutorAllocationRequest req2 = TaskExecutorAllocationRequest.of(worker2, constraints, createJobMetadata(), 1);
TaskExecutorAllocationRequest req1 = TaskExecutorAllocationRequest.of(worker1, constraints, createJobMetadata(), 1, MantisJobDurationType.Perpetual);
TaskExecutorAllocationRequest req2 = TaskExecutorAllocationRequest.of(worker2, constraints, createJobMetadata(), 1, MantisJobDurationType.Perpetual);

// Upsert reservations
upsertReservation(resourceClusterActor, probe, key1, constraints, Set.of(req1), 1, BASE_INSTANT.toEpochMilli());
Expand Down Expand Up @@ -293,8 +294,8 @@ public void testReservationWithMultipleConstraintGroups() throws Exception {
WorkerId workerA = WorkerId.fromIdUnsafe("job-a-worker-0-1");
WorkerId workerB = WorkerId.fromIdUnsafe("job-b-worker-0-1");

TaskExecutorAllocationRequest reqA = TaskExecutorAllocationRequest.of(workerA, constraintsA, createJobMetadata(), 1);
TaskExecutorAllocationRequest reqB = TaskExecutorAllocationRequest.of(workerB, constraintsB, createJobMetadata(), 1);
TaskExecutorAllocationRequest reqA = TaskExecutorAllocationRequest.of(workerA, constraintsA, createJobMetadata(), 1, MantisJobDurationType.Perpetual);
TaskExecutorAllocationRequest reqB = TaskExecutorAllocationRequest.of(workerB, constraintsB, createJobMetadata(), 1, MantisJobDurationType.Perpetual);

upsertReservation(resourceClusterActor, probe, key1, constraintsA, Set.of(reqA), 1, BASE_INSTANT.toEpochMilli());
upsertReservation(resourceClusterActor, probe, key2, constraintsB, Set.of(reqB), 1, BASE_INSTANT.toEpochMilli());
Expand All @@ -317,7 +318,7 @@ public void testReservationCancellation() throws Exception {
ReservationKey key = ReservationKey.builder().jobId("job-cancel").stageNumber(1).build();
SchedulingConstraints constraints = SchedulingConstraints.of(MACHINE, Optional.empty(), ImmutableMap.of());
WorkerId worker = WorkerId.fromIdUnsafe("job-cancel-worker-0-1");
TaskExecutorAllocationRequest req = TaskExecutorAllocationRequest.of(worker, constraints, createJobMetadata(), 1);
TaskExecutorAllocationRequest req = TaskExecutorAllocationRequest.of(worker, constraints, createJobMetadata(), 1, MantisJobDurationType.Perpetual);
resourceClusterActor.tell(MarkReady.INSTANCE, probe.getRef());
probe.expectMsg(Ack.getInstance());

Expand Down Expand Up @@ -374,7 +375,7 @@ public void testAssignmentRetry() throws Exception {
ReservationKey key = ReservationKey.builder().jobId("job-retry").stageNumber(1).build();
SchedulingConstraints constraints = SchedulingConstraints.of(MACHINE, Optional.empty(), ImmutableMap.of());
WorkerId worker = WorkerId.fromIdUnsafe("job-retry-worker-0-1");
TaskExecutorAllocationRequest req = TaskExecutorAllocationRequest.of(worker, constraints, createJobMetadata(), 1);
TaskExecutorAllocationRequest req = TaskExecutorAllocationRequest.of(worker, constraints, createJobMetadata(), 1, MantisJobDurationType.Perpetual);

upsertReservation(resourceClusterActor, probe, key, constraints, Set.of(req), 1, BASE_INSTANT.toEpochMilli());

Expand Down Expand Up @@ -426,7 +427,7 @@ public void testAssignmentFailure() throws Exception {
ReservationKey key = ReservationKey.builder().jobId("job-fail").stageNumber(1).build();
SchedulingConstraints constraints = SchedulingConstraints.of(MACHINE, Optional.empty(), ImmutableMap.of());
WorkerId worker = WorkerId.fromIdUnsafe("job-fail-worker-0-1");
TaskExecutorAllocationRequest req = TaskExecutorAllocationRequest.of(worker, constraints, createJobMetadata(), 1);
TaskExecutorAllocationRequest req = TaskExecutorAllocationRequest.of(worker, constraints, createJobMetadata(), 1, MantisJobDurationType.Perpetual);

upsertReservation(resourceClusterActor, probe, key, constraints, Set.of(req), 1, BASE_INSTANT.toEpochMilli());

Expand Down
3 changes: 2 additions & 1 deletion ...server/src/test/java/io/mantisrx/master/resourcecluster/ReservationRegistryActorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import java.util.Optional;
import java.util.stream.Collectors;
import io.mantisrx.runtime.MachineDefinition;
import io.mantisrx.runtime.MantisJobDurationType;
import io.mantisrx.server.core.scheduler.SchedulingConstraints;
import io.mantisrx.server.master.resourcecluster.ResourceCluster.NoResourceAvailableException;
import org.junit.AfterClass;
Expand Down Expand Up @@ -833,7 +834,7 @@ private static TaskExecutorAllocationRequest createAllocationRequest(
SchedulingConstraints constraints,
int stageNum
) {
return TaskExecutorAllocationRequest.of(workerId, constraints, null, stageNum);
return TaskExecutorAllocationRequest.of(workerId, constraints, null, stageNum, MantisJobDurationType.Perpetual);
}

@Test
Expand Down
3 changes: 2 additions & 1 deletion ...r/src/test/java/io/mantisrx/master/resourcecluster/ReservationScalerIntegrationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import io.mantisrx.master.resourcecluster.writable.ResourceClusterScaleRulesWritable;
import io.mantisrx.master.scheduler.CpuWeightedFitnessCalculator;
import io.mantisrx.runtime.MachineDefinition;
import io.mantisrx.runtime.MantisJobDurationType;
import io.mantisrx.server.core.TestingRpcService;
import io.mantisrx.server.core.domain.WorkerId;
import io.mantisrx.server.core.scheduler.SchedulingConstraints;
Expand Down Expand Up @@ -247,7 +248,7 @@ private void addReservation(ActorRef rcActor, String jobId, int stageNum, int nu
for (int i = 0; i < numWorkers; i++) {
WorkerId workerId = WorkerId.fromIdUnsafe(jobId + "-worker-" + stageNum + "-" + i);
allocationRequests.add(TaskExecutorAllocationRequest.of(
workerId, constraints, null, stageNum));
workerId, constraints, null, stageNum, MantisJobDurationType.Perpetual));
}

UpsertReservation upsert = UpsertReservation.builder()
Expand Down
13 changes: 7 additions & 6 deletions ...ontrol-plane-server/src/test/java/io/mantisrx/master/resourcecluster/ReservationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

import static io.mantisrx.server.master.resourcecluster.proto.MantisResourceClusterReservationProto.*;
import io.mantisrx.runtime.MachineDefinition;
import io.mantisrx.runtime.MantisJobDurationType;
import io.mantisrx.server.core.scheduler.SchedulingConstraints;
import io.mantisrx.server.core.domain.WorkerId;
import io.mantisrx.server.master.resourcecluster.ClusterID;
Expand Down Expand Up @@ -149,13 +150,13 @@ public void testFromUpsertReservation_withActualAllocationRequests() {
// Create mock TaskExecutorAllocationRequest objects
// Note: In a real scenario, these would be properly constructed with all required fields
TaskExecutorAllocationRequest req1 = TaskExecutorAllocationRequest.of(
worker1, constraints, null, 1);
worker1, constraints, null, 1, MantisJobDurationType.Perpetual);

TaskExecutorAllocationRequest req2 = TaskExecutorAllocationRequest.of(
worker2, constraints, null, 1);
worker2, constraints, null, 1, MantisJobDurationType.Perpetual);

TaskExecutorAllocationRequest req3 = TaskExecutorAllocationRequest.of(
worker3, constraints, null, 1);
worker3, constraints, null, 1, MantisJobDurationType.Perpetual);

Set<TaskExecutorAllocationRequest> allocationRequests = Set.of(req1, req2, req3);
Set<WorkerId> expectedWorkerIds = Set.of(worker1, worker2, worker3);
Expand Down Expand Up @@ -305,13 +306,13 @@ public void testFromUpsertReservation_withDuplicateWorkerIds() {

// Create allocation requests with duplicate WorkerIds
TaskExecutorAllocationRequest req1 = TaskExecutorAllocationRequest.of(
worker1, constraints, null, 2);
worker1, constraints, null, 2, MantisJobDurationType.Perpetual);

TaskExecutorAllocationRequest req2 = TaskExecutorAllocationRequest.of(
worker2, constraints, null, 2);
worker2, constraints, null, 2, MantisJobDurationType.Perpetual);

TaskExecutorAllocationRequest req3 = TaskExecutorAllocationRequest.of(
worker1, constraints, null, 2); // Duplicate worker1
worker1, constraints, null, 2, MantisJobDurationType.Perpetual); // Duplicate worker1

Set<TaskExecutorAllocationRequest> allocationRequests = new HashSet<>();
allocationRequests.add(req1);
Expand Down
29 changes: 15 additions & 14 deletions ...est/java/io/mantisrx/master/resourcecluster/ResourceClusterActorClusterUsageAkkaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import io.mantisrx.master.resourcecluster.proto.GetClusterUsageResponse.UsageByGroupKey;
import io.mantisrx.master.scheduler.CpuWeightedFitnessCalculator;
import io.mantisrx.runtime.MachineDefinition;
import io.mantisrx.runtime.MantisJobDurationType;
import io.mantisrx.server.core.TestingRpcService;
import io.mantisrx.server.core.domain.WorkerId;
import io.mantisrx.server.core.scheduler.SchedulingConstraints;
Expand Down Expand Up @@ -214,7 +215,7 @@ public void testGetTaskExecutorsUsage_WithAllocationAttributes() throws Exceptio
}

// reserve jdk 17 TE and check usage
Set<TaskExecutorAllocationRequest> requests = Collections.singleton(TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 0));
Set<TaskExecutorAllocationRequest> requests = Collections.singleton(TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 0, MantisJobDurationType.Perpetual));
assertEquals(
TASK_EXECUTOR_ID_3,
resourceCluster.getTaskExecutorsFor(requests).get().values().stream().findFirst().get());
Expand All @@ -235,9 +236,9 @@ public void testGetTaskExecutorsUsage_WithAllocationAttributes() throws Exceptio
public void testGetTaskExecutorsUsage_WithAllocationAttributesWithPendingJobSingleStage() throws Exception {
// Requesting 3 workers but only 1 available and satisfying constraints --> add job to pending cache
Set<TaskExecutorAllocationRequest> requests = ImmutableSet.of(
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 0),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-2"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 0),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-3"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 0));
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 0, MantisJobDurationType.Perpetual),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-2"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 0, MantisJobDurationType.Perpetual),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-3"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 0, MantisJobDurationType.Perpetual));
assertThrows(ExecutionException.class, () -> resourceCluster.getTaskExecutorsFor(requests).get());

// Test get cluster usage
Expand All @@ -258,9 +259,9 @@ public void testGetTaskExecutorsUsage_WithAllocationAttributesWithPendingJobSing
public void testGetTaskExecutorsUsage_WithAllocationAttributesWithPendingJobMultiStageOnlyOneAvailable() throws Exception {
// Requesting 3 workers: 1 small & 2 medium but only 1 small & 1 medium available and satisfying constraints --> add job to pending cache
Set<TaskExecutorAllocationRequest> requests = ImmutableSet.of(
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_1, ImmutableMap.of("jdk", "17")), null, 0),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-2"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 1),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-3"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 1));
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_1, ImmutableMap.of("jdk", "17")), null, 0, MantisJobDurationType.Perpetual),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-2"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 1, MantisJobDurationType.Perpetual),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-3"), SchedulingConstraints.of(MACHINE_DEFINITION_2, ImmutableMap.of("jdk", "17")), null, 1, MantisJobDurationType.Perpetual));
assertThrows(ExecutionException.class, () -> resourceCluster.getTaskExecutorsFor(requests).get());

// Test get cluster usage
Expand Down Expand Up @@ -340,9 +341,9 @@ public void testGetTaskExecutorsUsage_WithSizeNameAndAllocation() throws Excepti

// Requesting 3 workers with size name: 1 small & 2 large --> all available will decrement count as well
Set<TaskExecutorAllocationRequest> requests = ImmutableSet.of(
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_1, Optional.of("small"), ImmutableMap.of("jdk", "17")), null, 0),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-2"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-3"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1));
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_1, Optional.of("small"), ImmutableMap.of("jdk", "17")), null, 0, MantisJobDurationType.Perpetual),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-2"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1, MantisJobDurationType.Perpetual),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-3"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1, MantisJobDurationType.Perpetual));
assertEquals(
ImmutableSet.of("taskExecutorId4", "taskExecutorId5", "taskExecutorId6"),
new HashSet<>(resourceCluster.getTaskExecutorsFor(requests).get().values().stream().map(TaskExecutorID::getResourceId).collect(Collectors.toList())));
Expand All @@ -368,10 +369,10 @@ public void testGetTaskExecutorsUsage_WithSizeNameAndPending() throws Exception

// Requesting 3 workers with size name: 1 small & 3 large --> cannot find enough large available -> add to pending
Set<TaskExecutorAllocationRequest> requests = ImmutableSet.of(
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_1, Optional.of("small"), ImmutableMap.of("jdk", "17")), null, 0),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-2"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-3"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-4"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1));
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-0-1"), SchedulingConstraints.of(MACHINE_DEFINITION_1, Optional.of("small"), ImmutableMap.of("jdk", "17")), null, 0, MantisJobDurationType.Perpetual),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-2"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1, MantisJobDurationType.Perpetual),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-3"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1, MantisJobDurationType.Perpetual),
TaskExecutorAllocationRequest.of(WorkerId.fromIdUnsafe("late-sine-function-tutorial-1-worker-1-4"), SchedulingConstraints.of(MACHINE_DEFINITION_2, Optional.of("large"), ImmutableMap.of("jdk", "17")), null, 1, MantisJobDurationType.Perpetual));
assertThrows(ExecutionException.class, () -> resourceCluster.getTaskExecutorsFor(requests).get());

// Test get cluster usage
Expand Down
Loading
Loading