### Question:
ProductEventConsumer extends AbstractKafkaConsumer { @KafkaListener(topics = "${eventing.topic_name}") public void listen(final ConsumerRecord<String, String> consumerRecord, final Acknowledgment ack) { super.handleConsumerRecord(consumerRecord, ack); } @Inject protected  ProductEventConsumer(ProductEventProcessor messageProcessor, UnprocessableEventService unprocessableEventService); @KafkaListener(topics = "${eventing.topic_name}") void listen(final ConsumerRecord<String, String> consumerRecord, final Acknowledgment ack);  }### Answer:
@Test public void eventWithSyntaxErrorShouldNeitherBeProcessedNorStoredAsUnprocessable() { productEventConsumer = new ProductEventConsumer(mockedProcessor(SUCCESS), unprocessableEventService); productEventConsumer.listen(CONSUMER_RECORD,ack); verify(ack).acknowledge(); verify(unprocessableEventService, never()).save(any()); }
@Test public void eventLeadingToUnexpectedErrorShouldBeStoredAsUnprocessable() { final ProductEventProcessor processor = mockedProcessor(UNEXPECTED_ERROR); productEventConsumer = new ProductEventConsumer(processor, unprocessableEventService); productEventConsumer.listen(CONSUMER_RECORD,ack); verify(ack).acknowledge(); verify(processor).processConsumerRecord(any()); verify(unprocessableEventService).save(any()); }
@Test(expected = TemporaryKafkaProcessingError.class) public void temporaryErrorShouldStoreEventAsUnprocessable() { final ProductEventProcessor processor = mockedProcessor(TEMPORARY_ERROR); productEventConsumer = new ProductEventConsumer(processor, unprocessableEventService); productEventConsumer.listen(CONSUMER_RECORD,ack); verify(ack, never()).acknowledge(); verify(processor).processEvent(any()); verify(unprocessableEventService).save(any()); }