I am having problems with Mockito asynchronous verification

[Jim McMaster]

I am trying to write an integration test for an asynchronous task.  I just updated to Mockito 5.7.0.

The code sends a Kafka message, then verifies a method has been called on a mock TaskManager instance.  The Kafka message is directed to a Runnable.  I am verifying with a timeout to allow Kafka to deliver the message.  The test passes 100% of the time in my local IntelliJ Idea environment, but fails in our Jenkins environment.

The Runnable in question does the following:

@Override
public void run() {
  logger.info("DatFileUploadEvent received {}", LocalTime.now());
  DatFileUploadEvent uploadEvent = event.getPayload();
  if (!uploadEvent.getDataType().equals(DataType.SALES)) {
    return;
  }
  String businessId = uploadEvent.getBusinessId();
  InstallInformation installInformation =
    ingestClient.getInstallInformation(List.of(businessId)).get(businessId);
  repository.writeDataMonitorResult(
      DataMonitorResult.newBuilder()
          .setBusinessId(businessId)
          .setBusinessDay(uploadEvent.getBusinessDay())
          .setPosId(installInformation.getPosId())
          .setTimeStamp(DateTimeUtils.fromInstant(clock.instant()))
          .setSalesStatus(DATA_MONITOR_NOT_RUN)
          .setLaborStatus(DATA_MONITOR_NOT_SCHEDULED)
          .build());
  taskManager.scheduleDatFileMonitorTask(dataTaskFactory.create(businessId));
  logger.info("DatFileUploadEvent processed {}", LocalTime.now());
}

The test is:

@Test
public void datFileUploadEvent() {
  DataMonitorTask task =
      new DataMonitorTask(null, null, null, businessClient, null, BUSINESS_ID_1);
  InstallInformation information =       InstallInformation.newBuilder().setPosId("PosId").build();
  when(dataMonitorTaskFactory.create(BUSINESS_ID_1)).thenReturn(task);
  when(ingestClient.getInstallInformation(List.of(BUSINESS_ID_1)))
.thenReturn(Map.of(BUSINESS_ID_1, information));

  kafkaEventSender.sendRecord(
      new KafkaEvent<>(
          DatFileUploadEvent.newBuilder()
              .setBusinessId(BUSINESS_ID_1)
              .setBusinessDay(BusinessDay.newBuilder().setValue("20231030"))
              .setDataType(SALES)
              .build(),
          "Actor",
          "AggregateType",
          "AggregateId"),
      List.of(DatFileUploadEvent.getDescriptor().getFullName()));
  logger.info("DatFileUploadEvent sent {}", LocalTime.now());
  verify(taskManager, timeout(5000).atLeast(1)).scheduleDatFileMonitorTask(task);
  logger.info("DatFileUploadEvent verified {}", LocalTime.now());
}

I enabled verbose logging on the mock taskManager, and here is the output I get in Jenkins:

------------------------------------------------------

17:20:08.335 [main] INFO com.averoinc.ingest.kafka.consumer.KafkaEventWorkerIT - DatFileUploadEvent sent 17:20:08.335705 17:20:08.406 [Kafka-Consumer-1] INFO com.averoinc.kafka.client.AveroKafkaConsumer - Work is complete, unpausing 20 partitions on topic Ingest and committing work 17:20:08.844 [Kafka-Consumer-0] INFO com.averoinc.kafka.client.AveroKafkaConsumer - Event received: Kafka Details="ConsumerRecord(topic = avero.ftcs.DatFileUploadEvent, partition = 35, leaderEpoch = 0, offset = 0, CreateTime = 1699291208328, serialized key size = 11, serialized value size = 126, headers = RecordHeaders(headers = [], isReadOnly = false), key = AggregateId, value = [B@18bcb3ef)", type=avero.ftcs.DatFileUploadEvent, actor=Actor, id=, aggregateId=AggregateId 17:20:08.846 [pool-15-thread-1] INFO com.averoinc.ingest.kafka.consumer.DatFileUploadWorker - DatFileUploadEvent received 17:20:08.846575 17:20:08.846 [Kafka-Consumer-0] INFO com.averoinc.kafka.client.AveroKafkaConsumer - There is work in progress, pausing 20 partitions on topic avero.ftcs.DatFileUploadEvent 17:20:08.861 [pool-15-thread-1] INFO com.averoinc.ingest.kafka.consumer.DatFileUploadWorker - DatFileUploadEvent processed 17:20:08.861568 17:20:08.952 [Kafka-Consumer-0] INFO com.averoinc.kafka.client.AveroKafkaConsumer - Work is complete, unpausing 20 partitions on topic avero.ftcs.DatFileUploadEvent and committing work ############ Logging method invocation #8 on mock/spy ######## taskManager.scheduleDatFileMonitorTask( com.averoinc.ingest.tasks.DataMonitorTask@87ea043 ); invoked: -> at com.averoinc.ingest.kafka.consumer.KafkaEventWorkerIT.datFileUploadEvent(KafkaEventWorkerIT.java:215) has thrown: class org.mockito.exceptions.verification.WantedButNotInvoked with message Wanted but not invoked: taskManager.scheduleDatFileMonitorTask( com.averoinc.ingest.tasks.DataMonitorTask@87ea043 ); -> at com.averoinc.ingest.tasks.TaskManager.scheduleDatFileMonitorTask(TaskManager.java:180) Actually, there were zero interactions with this mock.

------------------------------------

The logging for Invocation 8 appears to show the method being invoked, and then the verify immediately says it was not invoked.  All this happens well within the five-second timeout.

I have been beating my head against this for three days.  Can anyone tell me what I am doing wrong?  Or is there a problem with Mockito?

--

Jim McMaster