我想设置一些关于Kafka的测试。我的Kafka通话设置如下:
@Async // allows function to be async. the config also has to be set
@Override
public void publishEventToTopic() {
ListenableFuture<SendResult<String, KafkaRequest>> future = kafkaTemplate.send(topic, request);
future.addCallback(new ListenableFutureCallback<SendResult<String, KafkaRequest>>() {
@Override
public void onSuccess(SendResult<String, KafkaRequest> result) {
log.info("Success");
}
@Override
public void onFailure(Throwable ex) {
log.info("Failure");
}
});
}
我想做的是测试两个场景(onsuccess和onfailure)。我试着遵循这个问题,但我遇到了一个错误,它说,“主题不能为空”。以下是我尝试的:
@Test
public void test2() {
String key = "test1";
String topic = "sender.t";
long offset = 1;
int partition = 0;
SendResult<String, Object> sendResult = mock(SendResult.class);
ListenableFuture<SendResult<String, KafkaRequest>> responseFuture = mock(ListenableFuture.class);
RecordMetadata recordMetadata = new RecordMetadata(new TopicPartition(topic, partition), offset, 0L, 0L, 0L, 0, 0);
ProducerRecord producerRecord = new ProducerRecord(topic, partition, key, "");
given(sendResult.getRecordMetadata()).willReturn(recordMetadata);
given(sendResult.getProducerRecord()).willReturn(producerRecord);
when(template.send(any(), any())).thenReturn(responseFuture);
doAnswer(invocationOnMock -> {
ListenableFutureCallback listenableFutureCallback = invocationOnMock.getArgument(0);
listenableFutureCallback.onFailure(throwable);
return null;
})
.when(responseFuture).addCallback(any()); //this is the line that is throwing an error when trying to debug.
publisher.publishEventToTopic();
}
我不太清楚这意味着什么,所以几个小时后,我决定尝试一种不同的方法-使用mockproducer类。它似乎奏效了,但并不始终如一:
@Test
public void test3() throws InterruptedException {
ListenableFuture<SendResult<String, KafkaRequest>> responseErrorFuture = mock(ListenableFuture.class);
MockProducer mockProducer = new MockProducer(Cluster.empty(), false, null, null, null);
mockProducer.clear();
mockProducer.errorNext(new RuntimeException("Offset commit failed on partition my-topic-2-9 at offset 0")); //this doesn't work anymore.
publisher.publishEventToTopic();
}
所以我试着加上这个,但它抱怨我没有嘲笑正确的东西:
doAnswer(invocation -> {
ListenableFutureCallback listenableFutureCallback = invocation.getArgument(0);
mockProducer.errorNext(new RuntimeException("Offset commit failed on partition my-topic-2-9 at offset 0"));
return null;
}).when(mockProducer).send(any(), any());
我的问题是,有人知道我在发送kafka请求时测试回调的onfailure方法有什么错吗?
谢谢!
暂无答案!
目前还没有任何答案,快来回答吧!