recommit kafka-1112 since it's inadvently reverted in the commit of kafka-1135