Модульное юнит-тестирование с Kafka - callback mockproducer - PullRequest
Новая
       46

Модульное юнит-тестирование с Kafka - callback mockproducer

1 голос
/ 30 марта 2020

Я хотел бы настроить некоторые тесты вокруг вызова Кафки. У меня есть следующая настройка обратного вызова с моим вызовом 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");
            }
        });
}

Что я хотел бы сделать, это протестировать оба сценария ios (onSuccess & onFailure). Я пытался ответить на этот ТАК вопрос , но я сталкиваюсь с ошибкой, в которой говорится, что "topi c не может быть нулевым". Вот что я попробовал: 

@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());

Мой вопрос Кто-нибудь имеет представление о том, что я делаю неправильно, чтобы не протестировать метод обратного вызова onFailure при отправке запроса Kafka?

Спасибо!

...