I have a method for sending kafka message like this:
@Async
public void sendMessage(String topicName, Message message) {
ListenableFuture<SendResult<String, Message >> future = kafkaTemplate.send(topicName, message);
future.addCallback(new ListenableFutureCallback<>() {
@Override
public void onSuccess(SendResult<String, Message > result) {
//do nothing
}
@Override
public void onFailure(Throwable ex) {
log.error("something wrong happened"!);
}
});
}
And now I am writing unit tests for it. I would like to test also the two callback methods onSuccess
and onFailure
methods, so my I idea is to mock the KafkaTemplate, something like :
KafkaTemplate kafkaTemplate = Mockito.mock(KafkaTemplate.class);
But now I am getting stuck on the mocking result for these two cases:
when(kafkaTemplate.send(anyString(), any(Message.class))).thenReturn(????);
what should I put in the thenReturn
method for the case success and for the case failure? Does anyone have an idea please? Thank you very much!
You can mock the template but it's better to mock the interface. Sender sender = new Sender(); KafkaOperations template = mock(KafkaOperations. class); SettableListenableFuture<SendResult<String, String>> future = new SettableListenableFuture<>(); when(template. send(anyString(), any(Message.
Testing a Kafka Consumer Consuming data from Kafka consists of two main steps. Firstly, we have to subscribe to topics or assign topic partitions manually. Secondly, we poll batches of records using the poll method. The polling is usually done in an infinite loop.
If you want to write integration tests using EmbeddedKafka , then you can do something like this. Assume we have some KafkaListener , which accepts a RequestDto as a Payload . In your test class you should create a TestConfiguration in order to create producer beans and to autowire KafkaTemplate into your test.
Spring Kafka Test is a Java Archive File that contains some helpful utilities to test your application.
You can mock the template but it's better to mock the interface.
Sender sender = new Sender();
KafkaOperations template = mock(KafkaOperations.class);
SettableListenableFuture<SendResult<String, String>> future = new SettableListenableFuture<>();
when(template.send(anyString(), any(Message.class))).thenReturn(future);
sender.setTemplate(template);
sender.send(...);
future.set(new SendResult<>(...));
...or...
future.setException(...
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With