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