Auto recover when disconnected from kafka