close socket on exception
reset iterators; catch exceptions; detect when kafka is lying (i.e. returning 0 bytes but there is data left in the buffer)