Fix the issue of getting initial offset with user provided start

[chaoqin-li1123]

Motivation

Fix the 2 bugs in getUserProvidedMessageId

1. consumer.asInstanceOf[ConsumerImpl[GenericRecord]].hasMessageAvailable may throw exception if consumer is freshly started.
2. initial message id is not wrap in UserProvidedMessageId class, which cause the first message to be skipped.
3. when consumer seek() and read() it actually skip the start message if the message is not a batch message

Modifications

No longer call hasMessageAvailable() method.
Wrap initial offset inside UserProvidedMessageId
change to use reader instead of consumer.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

• This change is a trivial rework / code cleanup without any test coverage.

• This change is already covered by existing tests, such as:

• This change added tests and can be verified as follows:

Documentation

Check the box below.

Need to update docs?

• doc-required
• no-need-doc
• doc

[nlu90]

This call seems weird to me. Shouldn't the seek method also handle reconnecting?

[chaoqin-li1123]

I find another issue, when consumer seek on offset that is not a batch message id, it will skip a message, previously this method was using reader instead of consumer, which doesn't have the same issue. I have verified this in the test.