[improve][pulsar-io-kafka] Add option to copy Kafka headers to Pulsar properties

[aymkhalil]

Fixes #17760

Motivation

Users of Kafka source today can't access their Kafka headers in Pulsar. This patch adds a config to enable this copy to take place in addition to some standard, useful Kafka headers.

Modifications

• A new KafkaSourceConfig flag is added: copyHeadersEnabled of type boolean and default value of false
• If the flag is toggle on, the following Kafka headers will be copied over to Pulsars properties.
• __kafka_topic : the name of the Kafka topic consumed by the source.
• __kafka_partition: the number of paritioed on the consumed kafka topics
• __kafka_offset: the current offset of the kafka message currently consumed by the source
• User defined header: The customer headers a Kafka producer could generated as pairs of <String, byte[]>. Since Pulsar properties are of type <String, String>, the Kafka byte[] values are based64 encoded.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added unit tests to check the Kafka source respects the configs and generated the expected Pulsar properties map
• Extended integration test for Avro Kafka source to generate headers and verify them on Pulsar side
  Update : The integ test is blocked by kafka-avro-console-producer not respecting --property parse.headers=true confluentinc/schema-registry#2390 and will be disabled for now.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (add or upgrade a dependency) (no)
• The public API (no)
• The schema (no)
• The default values of configurations (no)
• The binary protocol (no)
• The REST endpoints (no)
• The admin CLI options (no)
• Anything that affects deployment (no)

Documentation

• [] doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  Will be generated automatically

• doc
  (Your PR contains doc changes)

• doc-complete
  (Docs have been already added)

Matching PR in forked repository

PR in forked repository: aymkhalil#1

[eolivelli]

LGTM

[eolivelli]

There is no need to add docs, as docs are automatically generated

[nicoloboschi]

Lgtm