@Test
public void sendMultipleMessagesWithFailingMessageWrite(TestContext context) throws ValidationException {

    Async async = context.async();
    String topic = "myTopic";
    final List<KafkaProducerRecord<String, String>> records =
            buildRecords(topic, Buffer.buffer(buildThreeRecordsPayload("key_1", "key_2", "key_3").encode()));
    doAnswer(invocation -> {
        Handler<AsyncResult<RecordMetadata>> handler = (Handler<AsyncResult<RecordMetadata>>) invocation.getArguments()[1];
        handler.handle(Future.succeededFuture(new RecordMetadata(1,1,1,1, topic)));
        return null;
    }).when(producer).write(any(), any());
    doAnswer(invocation -> {
        Handler<AsyncResult<RecordMetadata>> handler = (Handler<AsyncResult<RecordMetadata>>) invocation.getArguments()[1];
        handler.handle(Future.failedFuture("Message with key '" + records.get(1).key() + "' failed."));
        return null;
    }).when(producer).write(eq(records.get(1)), any());
    kafkaMessageSender.sendMessages(producer, records).setHandler(event -> {
        context.assertFalse(event.succeeded());
        context.assertEquals("Message with key 'key_2' failed.", event.cause().getMessage());
        async.complete();
    });
    ArgumentCaptor<KafkaProducerRecord> recordCaptor = ArgumentCaptor.forClass(KafkaProducerRecord.class);
    Mockito.verify(producer, times(2)).write(recordCaptor.capture(), any());
    // verify only the first two messages was sent
    context.assertEquals(2, recordCaptor.getAllValues().size());
    context.assertEquals(records.get(0), recordCaptor.getAllValues().get(0));
    context.assertEquals(records.get(1), recordCaptor.getAllValues().get(1));
}