diff --git a/kafka/kafkatest/message_consumer.go b/kafka/kafkatest/message_consumer.go
index <HASH>..<HASH> 100644
--- a/kafka/kafkatest/message_consumer.go
+++ b/kafka/kafkatest/message_consumer.go
@@ -23,6 +23,12 @@ func (consumer *MessageConsumer) Incoming() chan kafka.Message {
 	return consumer.messages
 }
 
+// CommitAndRelease commits the message, releases the listener to consume next
+func (consumer *MessageConsumer) CommitAndRelease(m kafka.Message) {
+	m.Commit()
+	return
+}
+
 // Closer returns the stored closer channel.
 func (consumer *MessageConsumer) Closer() chan bool {
 	return consumer.closer
add CommitAndRelease to kafkatest