### Question:
ToKafkaSenderService implements IToKafkaSenderService { @Override @Transactional(rollbackFor = Exception.class) public void sendMessage(SendMessageRequest request) { ToKafkaMessages messages = new ToKafkaMessages().setTopic(request.getTopic()).add(convert(request)); taskService.addTask(new ITasksService.AddTaskRequest().setType(ToKafkaTaskType.VALUE).setSubType(messages.getTopic()).setData(messages) .setRunAfterTime(request.getSendAfterTime())); }  @Override @Transactional(rollbackFor = Exception.class) void sendMessage(SendMessageRequest request); @Override @Transactional(rollbackFor = Exception.class) void sendMessages(SendMessagesRequest request);  }### Answer:
@Test void payloadIsBeingConverted() throws Exception { when(objectMapper.writeValueAsString("abc")).thenReturn("abc"); toKafkaSenderService.sendMessage(new SendMessageRequest().setPayload("abc")); verify(objectMapper).writeValueAsString("abc"); verify(taskService).addTask(any()); }
@Test void payloadStringIsUsedAsItIs() throws Exception { toKafkaSenderService.sendMessage(new SendMessageRequest().setPayloadString("abc")); verify(objectMapper, never()).writeValueAsString(anyString()); verify(taskService).addTask(any()); }
@Test void headersAreAddedToTheTaskRequest() { Multimap<String, byte[]> headers = HashMultimap.create(); headers.put("a", "b".getBytes(UTF_8)); toKafkaSenderService.sendMessage(new SendMessageRequest() .setPayload("abc") .setHeaders(headers)); ArgumentCaptor<AddTaskRequest> taskCaptor = ArgumentCaptor.forClass(AddTaskRequest.class); verify(taskService).addTask(taskCaptor.capture()); AddTaskRequest actualRequest = taskCaptor.getValue(); assertHeadersEqual(headers, actualRequest); }