flink kafka consumer should support rate control

Now kafka connector does not support rate control at all, this may lead checkpoints to fail and expire if downstream processor is slow. Though this will not compromise the guarantee of flink's checkpoint, this will stop kafka consumer from comitting offsets since kafka consumer commits offsets in the callback of checkpoint complete. The committed offsets is important for monitoring purpose. So I think we need to support rate limit control in kafka consumer.