KafkaPublishTask extends WorkflowSystemTask { @Override public void start(Workflow workflow, Task task, WorkflowExecutor executor) { long taskStartMillis = Instant.now().toEpochMilli(); task.setWorkerId(config.getServerId()); Object request = task.getInputData().get(requestParameter); if (Objects.isNull(request)) { markTaskAsFailed(task, MISSING_REQUEST); return; } KafkaPublishTask.Input input = objectMapper.convertValue(request, KafkaPublishTask.Input.class); if (StringUtils.isBlank(input.getBootStrapServers())) { markTaskAsFailed(task, MISSING_BOOT_STRAP_SERVERS); return; } if (StringUtils.isBlank(input.getTopic())) { markTaskAsFailed(task, MISSING_KAFKA_TOPIC); return; } if (Objects.isNull(input.getValue())) { markTaskAsFailed(task, MISSING_KAFKA_VALUE); return; } try { Future<RecordMetadata> recordMetaDataFuture = kafkaPublish(input); try { recordMetaDataFuture.get(); task.setStatus(Task.Status.COMPLETED); long timeTakenToCompleteTask = Instant.now().toEpochMilli() - taskStartMillis; logger.debug("Published message {}, Time taken {}", input, timeTakenToCompleteTask); } catch (ExecutionException ec) { logger.error("Failed to invoke kafka task: {} - execution exception ", task.getTaskId(), ec); markTaskAsFailed(task, FAILED_TO_INVOKE + ec.getMessage()); } } catch (Exception e) { logger.error("Failed to invoke kafka task:{} for input {} - unknown exception", task.getTaskId(), input, e); markTaskAsFailed(task, FAILED_TO_INVOKE + e.getMessage()); } } @Inject  KafkaPublishTask(Configuration config, KafkaProducerManager clientManager, ObjectMapper objectMapper); @Override void start(Workflow workflow, Task task, WorkflowExecutor executor); @Override boolean execute(Workflow workflow, Task task, WorkflowExecutor executor); @Override void cancel(Workflow workflow, Task task, WorkflowExecutor executor); @Override boolean isAsync();  }
@Test public void missingRequest_Fail() { KafkaPublishTask kPublishTask = new KafkaPublishTask(new SystemPropertiesConfiguration(), new KafkaProducerManager(new SystemPropertiesConfiguration()), objectMapper); Task task = new Task(); kPublishTask.start(Mockito.mock(Workflow.class), task, Mockito.mock(WorkflowExecutor.class)); Assert.assertEquals(Task.Status.FAILED, task.getStatus()); }
@Test public void missingValue_Fail() { Task task = new Task(); KafkaPublishTask.Input input = new KafkaPublishTask.Input(); input.setBootStrapServers("localhost:9092"); input.setTopic("testTopic"); task.getInputData().put(KafkaPublishTask.REQUEST_PARAMETER_NAME, input); KafkaPublishTask kPublishTask = new KafkaPublishTask(new SystemPropertiesConfiguration(), new KafkaProducerManager(new SystemPropertiesConfiguration()), objectMapper); kPublishTask.start(Mockito.mock(Workflow.class), task, Mockito.mock(WorkflowExecutor.class)); Assert.assertEquals(Task.Status.FAILED, task.getStatus()); }
@Test public void missingBootStrapServers_Fail() { Task task = new Task(); KafkaPublishTask.Input input = new KafkaPublishTask.Input(); Map<String, Object> value = new HashMap<>(); input.setValue(value); input.setTopic("testTopic"); task.getInputData().put(KafkaPublishTask.REQUEST_PARAMETER_NAME, input); KafkaPublishTask kPublishTask = new KafkaPublishTask(new SystemPropertiesConfiguration(), new KafkaProducerManager(new SystemPropertiesConfiguration()), objectMapper); kPublishTask.start(Mockito.mock(Workflow.class), task, Mockito.mock(WorkflowExecutor.class)); Assert.assertEquals(Task.Status.FAILED, task.getStatus()); }
@Test public void kafkaPublishExecutionException_Fail() throws ExecutionException, InterruptedException { Task task = getTask(); KafkaProducerManager producerManager = Mockito.mock(KafkaProducerManager.class); KafkaPublishTask kPublishTask = new KafkaPublishTask(new SystemPropertiesConfiguration(), producerManager, objectMapper); Producer producer = Mockito.mock(Producer.class); Mockito.when(producerManager.getProducer(Mockito.any())).thenReturn(producer); Future publishingFuture = Mockito.mock(Future.class); Mockito.when(producer.send(Mockito.any())).thenReturn(publishingFuture); ExecutionException executionException = Mockito.mock(ExecutionException.class); Mockito.when(executionException.getMessage()).thenReturn("Execution exception"); Mockito.when(publishingFuture.get()).thenThrow(executionException); kPublishTask.start(Mockito.mock(Workflow.class), task, Mockito.mock(WorkflowExecutor.class)); Assert.assertEquals(Task.Status.FAILED, task.getStatus()); Assert.assertEquals("Failed to invoke kafka task due to: Execution exception", task.getReasonForIncompletion()); }
@Test public void kafkaPublishUnknownException_Fail() throws ExecutionException, InterruptedException { Task task = getTask(); KafkaProducerManager producerManager = Mockito.mock(KafkaProducerManager.class); KafkaPublishTask kPublishTask = new KafkaPublishTask(new SystemPropertiesConfiguration(), producerManager, objectMapper); Producer producer = Mockito.mock(Producer.class); Mockito.when(producerManager.getProducer(Mockito.any())).thenReturn(producer); Mockito.when(producer.send(Mockito.any())).thenThrow(new RuntimeException("Unknown exception")); kPublishTask.start(Mockito.mock(Workflow.class), task, Mockito.mock(WorkflowExecutor.class)); Assert.assertEquals(Task.Status.FAILED, task.getStatus()); Assert.assertEquals("Failed to invoke kafka task due to: Unknown exception", task.getReasonForIncompletion()); }
@Test public void kafkaPublishSuccess_Completed() throws ExecutionException, InterruptedException { Task task = getTask(); KafkaProducerManager producerManager = Mockito.mock(KafkaProducerManager.class); KafkaPublishTask kPublishTask = new KafkaPublishTask(new SystemPropertiesConfiguration(), producerManager, objectMapper); Producer producer = Mockito.mock(Producer.class); Mockito.when(producerManager.getProducer(Mockito.any())).thenReturn(producer); Mockito.when(producer.send(Mockito.any())).thenReturn(Mockito.mock(Future.class)); kPublishTask.start(Mockito.mock(Workflow.class), task, Mockito.mock(WorkflowExecutor.class)); Assert.assertEquals(Task.Status.COMPLETED, task.getStatus()); }