KafkaPublishTaskMapper implements TaskMapper { @Override public List<Task> getMappedTasks(TaskMapperContext taskMapperContext) throws TerminateWorkflowException { logger.debug("TaskMapperContext {} in KafkaPublishTaskMapper", taskMapperContext); WorkflowTask taskToSchedule = taskMapperContext.getTaskToSchedule(); Workflow workflowInstance = taskMapperContext.getWorkflowInstance(); String taskId = taskMapperContext.getTaskId(); int retryCount = taskMapperContext.getRetryCount(); TaskDef taskDefinition = Optional.ofNullable(taskMapperContext.getTaskDefinition()) .orElseGet(() -> Optional.ofNullable(metadataDAO.getTaskDef(taskToSchedule.getName())) .orElse(null)); Map<String, Object> input = parametersUtils.getTaskInputV2(taskToSchedule.getInputParameters(), workflowInstance, taskId, taskDefinition); Task kafkaPublishTask = new Task(); kafkaPublishTask.setTaskType(taskToSchedule.getType()); kafkaPublishTask.setTaskDefName(taskToSchedule.getName()); kafkaPublishTask.setReferenceTaskName(taskToSchedule.getTaskReferenceName()); kafkaPublishTask.setWorkflowInstanceId(workflowInstance.getWorkflowId()); kafkaPublishTask.setWorkflowType(workflowInstance.getWorkflowName()); kafkaPublishTask.setCorrelationId(workflowInstance.getCorrelationId()); kafkaPublishTask.setScheduledTime(System.currentTimeMillis()); kafkaPublishTask.setTaskId(taskId); kafkaPublishTask.setInputData(input); kafkaPublishTask.setStatus(Task.Status.SCHEDULED); kafkaPublishTask.setRetryCount(retryCount); kafkaPublishTask.setCallbackAfterSeconds(taskToSchedule.getStartDelay()); kafkaPublishTask.setWorkflowTask(taskToSchedule); kafkaPublishTask.setWorkflowPriority(workflowInstance.getPriority()); if (Objects.nonNull(taskDefinition)) { kafkaPublishTask.setExecutionNameSpace(taskDefinition.getExecutionNameSpace()); kafkaPublishTask.setIsolationGroupId(taskDefinition.getIsolationGroupId()); kafkaPublishTask.setRateLimitPerFrequency(taskDefinition.getRateLimitPerFrequency()); kafkaPublishTask.setRateLimitFrequencyInSeconds(taskDefinition.getRateLimitFrequencyInSeconds()); } return Collections.singletonList(kafkaPublishTask); } KafkaPublishTaskMapper(ParametersUtils parametersUtils, MetadataDAO metadataDAO); @Override List<Task> getMappedTasks(TaskMapperContext taskMapperContext); static final Logger logger; }
@Test public void getMappedTasks() { WorkflowTask taskToSchedule = new WorkflowTask(); taskToSchedule.setName("kafka_task"); taskToSchedule.setType(TaskType.KAFKA_PUBLISH.name()); taskToSchedule.setTaskDefinition(new TaskDef("kafka_task")); String taskId = IDGenerator.generate(); String retriedTaskId = IDGenerator.generate(); Workflow workflow = new Workflow(); WorkflowDef workflowDef = new WorkflowDef(); workflow.setWorkflowDefinition(workflowDef); TaskMapperContext taskMapperContext = TaskMapperContext.newBuilder() .withWorkflowDefinition(workflowDef) .withWorkflowInstance(workflow) .withTaskDefinition(new TaskDef()) .withTaskToSchedule(taskToSchedule) .withTaskInput(new HashMap<>()) .withRetryCount(0) .withRetryTaskId(retriedTaskId) .withTaskId(taskId) .build(); List<Task> mappedTasks = kafkaTaskMapper.getMappedTasks(taskMapperContext); assertEquals(1, mappedTasks.size()); assertEquals(TaskType.KAFKA_PUBLISH.name(), mappedTasks.get(0).getTaskType()); }
@Test public void getMappedTasks_WithoutTaskDef() { WorkflowTask taskToSchedule = new WorkflowTask(); taskToSchedule.setName("kafka_task"); taskToSchedule.setType(TaskType.KAFKA_PUBLISH.name()); String taskId = IDGenerator.generate(); String retriedTaskId = IDGenerator.generate(); Workflow workflow = new Workflow(); WorkflowDef workflowDef = new WorkflowDef(); workflow.setWorkflowDefinition(workflowDef); TaskDef taskdefinition = new TaskDef(); String testExecutionNameSpace = "testExecutionNameSpace"; taskdefinition.setExecutionNameSpace(testExecutionNameSpace); String testIsolationGroupId = "testIsolationGroupId"; taskdefinition.setIsolationGroupId(testIsolationGroupId); TaskMapperContext taskMapperContext = TaskMapperContext.newBuilder() .withWorkflowDefinition(workflowDef) .withWorkflowInstance(workflow) .withTaskDefinition(taskdefinition) .withTaskToSchedule(taskToSchedule) .withTaskInput(new HashMap<>()) .withRetryCount(0) .withRetryTaskId(retriedTaskId) .withTaskId(taskId) .build(); List<Task> mappedTasks = kafkaTaskMapper.getMappedTasks(taskMapperContext); assertEquals(1, mappedTasks.size()); assertEquals(TaskType.KAFKA_PUBLISH.name(), mappedTasks.get(0).getTaskType()); assertEquals(testExecutionNameSpace, mappedTasks.get(0).getExecutionNameSpace()); assertEquals(testIsolationGroupId, mappedTasks.get(0).getIsolationGroupId()); }