### Question:
ToKafkaSenderService implements IToKafkaSenderService { @Override @Transactional(rollbackFor = Exception.class) public void sendMessages(SendMessagesRequest request) { List<ToKafkaMessages.Message> messageStream = request .getMessages() .stream() .map(this::convert) .collect(Collectors.toList()); splitToBatches(messageStream, batchSizeMb * MEGABYTE_MULTIPLIER).forEach(it -> sendBatch(request, it)); }  @Override @Transactional(rollbackFor = Exception.class) void sendMessage(SendMessageRequest request); @Override @Transactional(rollbackFor = Exception.class) void sendMessages(SendMessagesRequest request);  }### Answer:
@Test void payloadIsBeingConvertedInCaseOfBatch() throws Exception { when(objectMapper.writeValueAsString("abc")).thenReturn("abc"); toKafkaSenderService.sendMessages(new SendMessagesRequest().add(new SendMessagesRequest.Message().setPayload("abc"))); verify(objectMapper).writeValueAsString("abc"); verify(taskService).addTask(any()); }
@Test void payloadStringIsBeingConvertedInCaseOfBatch() throws Exception { toKafkaSenderService.sendMessages(new SendMessagesRequest().add(new SendMessagesRequest.Message().setPayloadString("abc"))); verify(objectMapper, never()).writeValueAsString(anyString()); verify(taskService).addTask(any()); }
@Test void headersAreAddedToTheTaskRequestInCaseOfBatch() { Multimap<String, byte[]> headers = HashMultimap.create(); headers.put("a", "b".getBytes(UTF_8)); toKafkaSenderService.sendMessages(new SendMessagesRequest().add(new SendMessagesRequest.Message() .setPayloadString("abc") .setHeaders(headers))); ArgumentCaptor<AddTaskRequest> taskCaptor = ArgumentCaptor.forClass(AddTaskRequest.class); verify(taskService).addTask(taskCaptor.capture()); AddTaskRequest actualRequest = taskCaptor.getValue(); assertHeadersEqual(headers, actualRequest); }
@Test void messagesAreSplitToBatches() { long millis = System.currentTimeMillis(); SendMessagesRequest request = new SendMessagesRequest(); List<String> randomStrings = IntStream.range(1, 1025) .mapToObj(i -> RandomStringUtils.randomAlphabetic(1024)) .collect(Collectors.toList()); for (int i = 0; i < 50; i++) { Collections.shuffle(randomStrings); request.add(new SendMessagesRequest.Message().setPayloadString(String.join("", randomStrings))); } System.out.printf("Created 50 MB of random strings in %d ms.%n", System.currentTimeMillis() - millis); toKafkaSenderService.sendMessages(request); verify(taskService, atLeast(8)).addTask(any()); }
@Test void messagesThatAreBiggerThanBatchSizeAreStillBeingSent() { long millis = System.currentTimeMillis(); SendMessagesRequest request = new SendMessagesRequest(); String s1 = RandomStringUtils.randomAlphabetic(1024 * 1024 * 6); String s2 = RandomStringUtils.randomAlphabetic(1024 * 1024 * 6); request.add(new SendMessagesRequest.Message().setPayloadString(s1)); request.add(new SendMessagesRequest.Message().setPayloadString(s2)); System.out.printf("Created 2 random 6 MB strings in %d ms.%n", System.currentTimeMillis() - millis); toKafkaSenderService.sendMessages(request); verify(taskService, times(2)).addTask(any()); }