@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());
}