...

Besides, we sometimes wants to "peek" the incoming records, to do some testing, without affecting the offsets, like the "peek" method provided in many data structures. So, in this "peek" method, we won't increase the position offset in the partition. That means, under after peek, the next "poll"ed records will still include the records returned by `peek`. Under the `enable.auto.commit = true` (default setting) case, because the committed offsets are not incremented, so it won't be incremented, and in the next "poll", the returned data will still include the records returned by `peek`affect the committed offsets. That means, after the consumer restarted or rebalanced, the next poll will always start from the offset before operating peek methods. (of course if user manually commit the offsets, the offsets will be incremented)

So, we should will have a `consumer#peek()` to allow consumers to:

1. peek what records existed at broker side and no increasing the position offsets.
2. test if return exceptions when there is connection error existed between consumer and broker (or other exceptions will be thrown by "poll")

Public Interfaces

Add a `peek` method into `Consumer` interface

...

1. peek what records existed at broker side and no increasing the position offsets.
2. test if return exceptions when there is connection error existed between consumer and broker (or other exceptions will be thrown by "poll")

Compatibility, Deprecation, and Migration Plan

...