fix kafka consumer bug