KafkaMessageSender { Future<Void> sendMessages(KafkaProducer<String, String> kafkaProducer, List<KafkaProducerRecord<String, String>> messages){ Future<Void> future = Future.future(); Future<Void> f = Future.succeededFuture(); messages.stream().reduce(f, (f1, message) -> f1.compose(ignore -> KafkaMessageSender.this.sendMessage(kafkaProducer, message)), (voidFuture, voidFuture2) -> null ).setHandler(res -> { if(res.succeeded()) { future.complete(); } else { future.fail(res.cause()); } }); return future; }    }
@Test public void sendSingleMessage(TestContext context) throws ValidationException { Async async = context.async(); String topic = "myTopic"; final List<KafkaProducerRecord<String, String>> records = buildRecords(topic, Buffer.buffer(buildSingleRecordPayload("someKey").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(KafkaProducerRecord.class), any()); kafkaMessageSender.sendMessages(producer, records).setHandler(event -> { context.assertTrue(event.succeeded()); async.complete(); }); Mockito.verify(producer, times(1)).write(eq(records.get(0)), any()); }
@Test public void sendSingleMessageWithoutKey(TestContext context) throws ValidationException { Async async = context.async(); String topic = "myTopic"; final List<KafkaProducerRecord<String, String>> records = buildRecords(topic, Buffer.buffer(buildSingleRecordPayload(null).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(KafkaProducerRecord.class), any()); kafkaMessageSender.sendMessages(producer, records).setHandler(event -> { context.assertTrue(event.succeeded()); async.complete(); }); Mockito.verify(producer, times(1)).write(eq(records.get(0)), any()); }
@Test public void sendMultipleMessages(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(KafkaProducerRecord.class), any()); kafkaMessageSender.sendMessages(producer, records).setHandler(event -> { context.assertTrue(event.succeeded()); async.complete(); }); ArgumentCaptor<KafkaProducerRecord> recordCaptor = ArgumentCaptor.forClass(KafkaProducerRecord.class); Mockito.verify(producer, times(3)).write(recordCaptor.capture(), any()); context.assertEquals(3, recordCaptor.getAllValues().size()); context.assertEquals(records.get(0), recordCaptor.getAllValues().get(0)); context.assertEquals(records.get(1), recordCaptor.getAllValues().get(1)); context.assertEquals(records.get(2), recordCaptor.getAllValues().get(2)); }
@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()); context.assertEquals(2, recordCaptor.getAllValues().size()); context.assertEquals(records.get(0), recordCaptor.getAllValues().get(0)); context.assertEquals(records.get(1), recordCaptor.getAllValues().get(1)); }