@Test public void verifyDeDupTaskIds() { List<String> expectedIdList = new ArrayList<String>() {{ add("task1"); add("task2"); add("task3"); add("task4"); }}; List<Task> activeTasks = new ArrayList<>(); activeTasks.add(Task.newBuilder().setId("task1").build()); activeTasks.add(Task.newBuilder().setId("task2").build()); activeTasks.add(Task.newBuilder().setId("task3").build()); List<Task> archivedTasks = new ArrayList<>(); archivedTasks.add(Task.newBuilder().setId("task2").build()); archivedTasks.add(Task.newBuilder().setId("task3").build()); archivedTasks.add(Task.newBuilder().setId("task4").build()); List<Task> tasks = GatewayJobServiceGateway.deDupTasks(activeTasks, archivedTasks); assertThat(tasks.size()).isEqualTo(4); List<String> taskIds = tasks.stream().map(Task::getId).sorted().collect(Collectors.toList()); assertThat(taskIds).isEqualTo(expectedIdList); activeTasks = new ArrayList<>(); activeTasks.add(Task.newBuilder().setId("task1").build()); activeTasks.add(Task.newBuilder().setId("task2").build()); archivedTasks = new ArrayList<>(); archivedTasks.add(Task.newBuilder().setId("task3").build()); archivedTasks.add(Task.newBuilder().setId("task4").build()); tasks = GatewayJobServiceGateway.deDupTasks(activeTasks, archivedTasks); assertThat(tasks.size()).isEqualTo(4); taskIds = tasks.stream().map(Task::getId).collect(Collectors.toList()); Collections.sort(taskIds); assertThat(taskIds).isEqualTo(expectedIdList); }
@VisibleForTesting static List<Task> deDupTasks(List<Task> activeTasks, List<Task> archivedTasks) { Map<String, Task> archivedTasksMap = archivedTasks.stream().collect(Collectors.toMap(Task::getId, Function.identity())); List<Task> uniqueActiveTasks = activeTasks.stream().filter(activeTask -> { if (archivedTasksMap.containsKey(activeTask.getId())) { logger.warn("Duplicate Task detected (archived) {} - (active) {}", archivedTasksMap.get(activeTask.getId()), activeTask); return false; } return true; }).collect(Collectors.toList()); uniqueActiveTasks.addAll(archivedTasks); return uniqueActiveTasks; }
GatewayJobServiceGateway extends JobServiceGatewayDelegate { @VisibleForTesting static List<Task> deDupTasks(List<Task> activeTasks, List<Task> archivedTasks) { Map<String, Task> archivedTasksMap = archivedTasks.stream().collect(Collectors.toMap(Task::getId, Function.identity())); List<Task> uniqueActiveTasks = activeTasks.stream().filter(activeTask -> { if (archivedTasksMap.containsKey(activeTask.getId())) { logger.warn("Duplicate Task detected (archived) {} - (active) {}", archivedTasksMap.get(activeTask.getId()), activeTask); return false; } return true; }).collect(Collectors.toList()); uniqueActiveTasks.addAll(archivedTasks); return uniqueActiveTasks; } }
GatewayJobServiceGateway extends JobServiceGatewayDelegate { @VisibleForTesting static List<Task> deDupTasks(List<Task> activeTasks, List<Task> archivedTasks) { Map<String, Task> archivedTasksMap = archivedTasks.stream().collect(Collectors.toMap(Task::getId, Function.identity())); List<Task> uniqueActiveTasks = activeTasks.stream().filter(activeTask -> { if (archivedTasksMap.containsKey(activeTask.getId())) { logger.warn("Duplicate Task detected (archived) {} - (active) {}", archivedTasksMap.get(activeTask.getId()), activeTask); return false; } return true; }).collect(Collectors.toList()); uniqueActiveTasks.addAll(archivedTasks); return uniqueActiveTasks; } @Inject  GatewayJobServiceGateway(GrpcRequestConfiguration tunablesConfiguration,
                                    GatewayConfiguration gatewayConfiguration,
                                    JobManagerConfiguration jobManagerConfiguration,
                                    JobManagementServiceStub client,
                                    JobStore store,
                                    LogStorageInfo<com.netflix.titus.api.jobmanager.model.job.Task> logStorageInfo,
                                    TaskRelocationDataInjector taskRelocationDataInjector,
                                    NeedsMigrationQueryHandler needsMigrationQueryHandler,
                                    @Named(JOB_STRICT_SANITIZER) EntitySanitizer entitySanitizer,
                                    DisruptionBudgetSanitizer disruptionBudgetSanitizer,
                                    @Named(SECURITY_GROUPS_REQUIRED_FEATURE) Predicate<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> securityGroupsRequiredPredicate,
                                    @Named(ENVIRONMENT_VARIABLE_NAMES_STRICT_VALIDATION_FEATURE) Predicate<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> environmentVariableNamesStrictValidationPredicate,
                                    JobAssertions jobAssertions,
                                    AdmissionValidator<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> validator,
                                    AdmissionSanitizer<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> sanitizer,
                                    TitusRuntime titusRuntime); }
GatewayJobServiceGateway extends JobServiceGatewayDelegate { @VisibleForTesting static List<Task> deDupTasks(List<Task> activeTasks, List<Task> archivedTasks) { Map<String, Task> archivedTasksMap = archivedTasks.stream().collect(Collectors.toMap(Task::getId, Function.identity())); List<Task> uniqueActiveTasks = activeTasks.stream().filter(activeTask -> { if (archivedTasksMap.containsKey(activeTask.getId())) { logger.warn("Duplicate Task detected (archived) {} - (active) {}", archivedTasksMap.get(activeTask.getId()), activeTask); return false; } return true; }).collect(Collectors.toList()); uniqueActiveTasks.addAll(archivedTasks); return uniqueActiveTasks; } @Inject  GatewayJobServiceGateway(GrpcRequestConfiguration tunablesConfiguration,
                                    GatewayConfiguration gatewayConfiguration,
                                    JobManagerConfiguration jobManagerConfiguration,
                                    JobManagementServiceStub client,
                                    JobStore store,
                                    LogStorageInfo<com.netflix.titus.api.jobmanager.model.job.Task> logStorageInfo,
                                    TaskRelocationDataInjector taskRelocationDataInjector,
                                    NeedsMigrationQueryHandler needsMigrationQueryHandler,
                                    @Named(JOB_STRICT_SANITIZER) EntitySanitizer entitySanitizer,
                                    DisruptionBudgetSanitizer disruptionBudgetSanitizer,
                                    @Named(SECURITY_GROUPS_REQUIRED_FEATURE) Predicate<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> securityGroupsRequiredPredicate,
                                    @Named(ENVIRONMENT_VARIABLE_NAMES_STRICT_VALIDATION_FEATURE) Predicate<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> environmentVariableNamesStrictValidationPredicate,
                                    JobAssertions jobAssertions,
                                    AdmissionValidator<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> validator,
                                    AdmissionSanitizer<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> sanitizer,
                                    TitusRuntime titusRuntime); @Override Observable<Job> findJob(String jobId, CallMetadata callMetadata); @Override Observable<Task> findTask(String taskId, CallMetadata callMetadata); @Override Observable<TaskQueryResult> findTasks(TaskQuery taskQuery, CallMetadata callMetadata); }
GatewayJobServiceGateway extends JobServiceGatewayDelegate { @VisibleForTesting static List<Task> deDupTasks(List<Task> activeTasks, List<Task> archivedTasks) { Map<String, Task> archivedTasksMap = archivedTasks.stream().collect(Collectors.toMap(Task::getId, Function.identity())); List<Task> uniqueActiveTasks = activeTasks.stream().filter(activeTask -> { if (archivedTasksMap.containsKey(activeTask.getId())) { logger.warn("Duplicate Task detected (archived) {} - (active) {}", archivedTasksMap.get(activeTask.getId()), activeTask); return false; } return true; }).collect(Collectors.toList()); uniqueActiveTasks.addAll(archivedTasks); return uniqueActiveTasks; } @Inject  GatewayJobServiceGateway(GrpcRequestConfiguration tunablesConfiguration,
                                    GatewayConfiguration gatewayConfiguration,
                                    JobManagerConfiguration jobManagerConfiguration,
                                    JobManagementServiceStub client,
                                    JobStore store,
                                    LogStorageInfo<com.netflix.titus.api.jobmanager.model.job.Task> logStorageInfo,
                                    TaskRelocationDataInjector taskRelocationDataInjector,
                                    NeedsMigrationQueryHandler needsMigrationQueryHandler,
                                    @Named(JOB_STRICT_SANITIZER) EntitySanitizer entitySanitizer,
                                    DisruptionBudgetSanitizer disruptionBudgetSanitizer,
                                    @Named(SECURITY_GROUPS_REQUIRED_FEATURE) Predicate<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> securityGroupsRequiredPredicate,
                                    @Named(ENVIRONMENT_VARIABLE_NAMES_STRICT_VALIDATION_FEATURE) Predicate<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> environmentVariableNamesStrictValidationPredicate,
                                    JobAssertions jobAssertions,
                                    AdmissionValidator<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> validator,
                                    AdmissionSanitizer<com.netflix.titus.api.jobmanager.model.job.JobDescriptor> sanitizer,
                                    TitusRuntime titusRuntime); @Override Observable<Job> findJob(String jobId, CallMetadata callMetadata); @Override Observable<Task> findTask(String taskId, CallMetadata callMetadata); @Override Observable<TaskQueryResult> findTasks(TaskQuery taskQuery, CallMetadata callMetadata);  }