Mockito Unit Tests stubbing a method throws NullPointerException

I am writing some unit tests for the run() method in the below class

public class TopicProcessor implements Runnable {

     private final KafkaConfig config;
     private final String inTopic;
     private final String outTopic;
     private final String inBroker;
     private final String outBroker;
     private KafkaConsumer<Long, byte[]> kafkaConsumer;
     private KafkaProducer<Long, byte[]> kafkaProducer;


     public void setKafkaConsumer(KafkaConsumer<Long, byte[]> kafkaConsumer) {
         this.kafkaConsumer = kafkaConsumer;
     }

     public void setKafkaProducer(KafkaProducer<Long, byte[]> kafkaProducer) {
         this.kafkaProducer = kafkaProducer;
     }

     public TopicProcessor(String topicKey, KafkaConfig config) {
         this.config = config;
         Map<String, Map<String, String>> mapping = this.config.getMapping(topicKey);
         this.inTopic = mapping.get("in").get("topic");
         this.outTopic = mapping.get("out").get("topic");
         this.inBroker = mapping.get("in").get("broker");
         this.outBroker = mapping.get("out").get("broker");
     }


     @Override
     public void run() {
         long now;
         getKafkaConsumer().subscribe(Collections.singletonList(this.inTopic));
         while (true) {
             ConsumerRecords<Long, byte[]> records = getKafkaConsumer().poll(Duration.ofMillis(100));
             meterInfo("consumer.received." + this.inTopic, records.count());
             for (ConsumerRecord<Long, byte[]> record : records) {
                 now = currentTimeMillis();
                 histogram("consumer.latency." + this.inTopic, now - record.timestamp());
                 getKafkaProducer().send(new ProducerRecord<>(this.outTopic, record.key(), record.value()));
                 meterInfo("publisher.sent." + this.outTopic);
             }
         }
     }

Basically, In the run() method, I am subscribed to a topic, polled all records using KafkaConsumer.poll() method and for each record, I am sending the record using KafkaProducer.send() method.

I have to write unit tests for below scenario : -

1. Test that KafkaConsumer.poll() method called one time
2. Test that KafkaProducer.send() method called one time

Below is my Test method: -

@Test
public void test(){
    //Mocks
    KafkaConsumer<Long, byte[]> consumerMock = mock(KafkaConsumer.class);
    KafkaProducer<Long, byte[]> producerMock = mock(KafkaProducer.class);

    //test data
    Map<TopicPartition, List<ConsumerRecord<Long, byte[]>>> records = new LinkedHashMap<>();
records.put(new TopicPartition("topic", 0), new ArrayList<>());
    ConsumerRecords<Long, byte[]> consumerRecords = new ConsumerRecords<>(records);

    ProducerRecord<Long, byte[]> producer_record = new ProducerRecord<Long, byte[]>("topic", 1, 0L, 1L, new byte[]{'a'});

    //object creation to call the method
    KafkaConfig kafkaConfig = new KafkaConfig(yamlConfig);
    TopicProcessor topicProcessor = new TopicProcessor("pacing_record", kafkaConfig);

    //stubbing Kafka methods - poll and send
    when(consumerMock.poll(Duration.ofMillis(anyInt()))).thenReturn(consumerRecords);  
    doReturn("sent").when(producerMock.send(producer_record, null));

    //Injecting mocks thorugh setter method
    topicProcessor.setKafkaConsumer(consumerMock);
    topicProcessor.setKafkaProducer(producerMock);

    //calling a class method run() from above class to verify that kafka methods are called
    topicProcessor.run();

    verify(consumerMock, times(1)).poll(Duration.ofMillis(anyInt()));

}

Below line in the above test method throws NullPointerException

    when(kafkaConsumerMock.poll(Duration.ofMillis(anyInt()))).thenReturn(consumerRecords);

Anyone know how to correctly stub the method and verify it is called or not?

when(kafkaConsumerMock.poll(Duration.ofMillis(anyInt())))

This is an incorrect usage of Mockito. The any* methods should only be used as the direct parameters to the mocked method.

I think what you're trying to say here is "for any duration, I don't care how long it is". If so, you can express that as:

when(kafkaConsumerMock.poll(any(Duration.class))

or

when(kafkaConsumerMock.poll(any())

if there is no ambiguous overload.

Similarly, to verify:

verify(kafkaConsumerMock, times(1)).poll(any(Duration.class))