Q:

How to test if method with @KafkaListener is being called

I'm really struggling to write a test to check if my Kafka Consumer is being correctly called when messages are sent to it's designated topic.  
My consumer:
@Service
@Slf4j
@AllArgsConstructor(onConstructor = @__(@Autowired))
public class ProcessingConsumer {

  private AppService appService;

  @KafkaListener(
      topics = "${topic}",
      containerFactory = "processingConsumerContainerFactory")
  public void listen(ConsumerRecord<Key, Value> message, Acknowledgment ack) {
    try {
      appService.processMessage(message);
      ack.acknowledge();
    } catch (Throwable t) {
      log.error("error while processing message!",  t);
    }
  }
}

My consumer config:
@EnableKafka
@Configuration
public class ProcessingCosumerConfig {

  @Value("${spring.kafka.schema-registry-url}")
  private String schemaRegistryUrl;

  private KafkaProperties props;

  public ProcessingCosumerConfig(KafkaProperties kafkaProperties) {
    this.props = kafkaProperties;
  }

  public Map<String, Object> deserializerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(KafkaAvroDeserializerConfig. SPECIFIC_AVRO_READER_CONFIG, true);
    props.put(KafkaAvroDeserializerConfig. SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);

    return props;
  }

  private KafkaAvroDeserializer getKafkaAvroDeserializer(Boolean isKey) {
    KafkaAvroDeserializer kafkaAvroDeserializer = new KafkaAvroDeserializer();
    kafkaAvroDeserializer.configure(deserializerConfigs(), isKey);
    return kafkaAvroDeserializer;
  }

  private DefaultKafkaConsumerFactory consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(
        props.buildConsumerProperties(),
        getKafkaAvroDeserializer(true),
        getKafkaAvroDeserializer(false));
  }

  @Bean(name = "processingConsumerContainerFactory")
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Key, Value>>
      kafkaListenerContainerFactory() {

    ConcurrentKafkaListenerContainerFactory<Key, Value>
        factory = new ConcurrentKafkaListenerContainerFactory<>();

    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckOnError(false);
    factory.getContainerProperties().setAckMode(ContainerProperties. AckMode. MANUAL_IMMEDIATE);
    factory.setErrorHandler(new SeekToCurrentErrorHandler());
    return factory;
  }
}

Finally, my (wannabe) test:
@DirtiesContext
public class ProcessingConsumerTest extends BaseIntegrationTest{

  @Autowired private ProcessingProducerFixture processingProducer;
  @Autowired private ProcessingConsumer processingConsumer;
  @org.springframework.beans.factory.annotation. Value("${topic}")
  String topic;

  @Test
  public void consumer_shouldConsumeMessages_whenMessagesAreSent() throws Exception{
    Thread.sleep(1000);
    ProducerRecord<Key, Value> message = new ProducerRecord<>(topic, new Key("b"), new Value("a", "b", "c", "d"));
    processingProducer.send(message);
  }
}

And that's about it for all I have so far. 
I've tried checking if this approach gets to the consumer manually using debug and also even just putting simple prints there but the execution simply doesn't seems to get there.  Also, if it could be somehow called correctly by my tests, I have no idea what to do to actually assert it in the actual test. 

A:

Inject a mock AppService into the listener and verify its processMessage() was called.