fix kafka consumer bug

fix kafka consumer bug