@Test void payloadStringIsBeingConvertedInCaseOfBatch() throws Exception { toKafkaSenderService.sendMessages(new SendMessagesRequest().add(new SendMessagesRequest.Message().setPayloadString("abc"))); verify(objectMapper, never()).writeValueAsString(anyString()); verify(taskService).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);  }