fix kafka key NPE and consumer close exception

this pull request fixes:

1. null pointer exception when reading a kafka message whose key is null;
2. consumer exception when fetch thread has been interrupted