[FLINK-8354] Add KafkaDeserializationSchema that directly uses ConsumerRecord

[aljoscha]

This is a rebased and cleaned-up version of #6615 that only has the Consumer changes.

What is the purpose of the change

• Add a new deserialization schema that directly uses ConsumerRecord, thereby exposing all available Kafka functionality.

Brief change log

• First we do some cleanup on the Kafka code
• Then we introduce the new schema, the existing schema now implements the new schema, for backwards compatibility
• Consumers and tests are adapted to use the new schema

Verifying this change

• This is covered by existing tests

[flinkbot]

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Review Progress

• ❓ 1. The [description] looks good.
• ❓ 2. There is [consensus] that the contribution should go into to Flink.
• ❓ 3. Needs [attention] from.
• ❓ 4. The change fits into the overall [architecture].
• ❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
• @flinkbot approve all to approve all aspects
• @flinkbot approve-until architecture to approve everything until architecture
• @flinkbot attention @username1 [@username2 ..] to require somebody's attention
• @flinkbot disapprove architecture to remove an approval you gave earlier

[aljoscha]

cc @alexeyt820
cc @azagrebin

[aljoscha]

Merged!

[lamberken]

@aljoscha hi, the java file was placed at wrong package
KafkaDeserializationSchemaWrapper.java

[aljoscha]

@lamber-ken How do you mean?

[lamberken]

@aljoscha, hi, the mismatch between KafkaDeserializationSchemaWrapper class's package name and file's location. #7902

[aljoscha]

Now I see, yes, we need to fix this.