add ConsumerListener, fix errors in unit test