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