Support for logging MDC in the Kafka buffer

[dlvenable]

Description

This adds logging MDC to the kafka buffer to disambiguate it against the kafka sink. Entry points into the Buffer interface set the MDC value. Also, the threads which the kafka buffer directly creates will have MDC and also have a useful thread name.

This MDC is now available for both Data Prepper loggers and Apache Kafka loggers.

One limitation is that the MDC is not available for the KafkaProducer. This is because the internal Apache Kafka libraries create that thread and appear to have no way to influence it.

Issues Resolved

First part of #4126

I plan to follow up with more changes for the kafka sink in a subsequent PR.

Results

I changed the Log4j configuration to include the kafkaPluginType MDC value.

appender.console.layout.pattern = %d{ISO8601} [%t] [%X{kafkaPluginType}] %-5p %40C - %m%n

Here is a sample of the logs produced:

2024-02-15T09:38:29,884 [kafka-buffer-1] [buffer] INFO  org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-data-prepper-1, groupId=data-prepper] Notifying assignor about the new Assignment(partitions=[test-encrypted-0])
2024-02-15T09:38:29,884 [kafka-buffer-2] [buffer] INFO  org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-data-prepper-2, groupId=data-prepper] Adding newly assigned partitions:
2024-02-15T09:38:29,886 [kafka-buffer-1] [buffer] INFO  org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-data-prepper-1, groupId=data-prepper] Adding newly assigned partitions: test-encrypted-0
2024-02-15T09:38:29,886 [kafka-buffer-1] [buffer] INFO  org.opensearch.dataprepper.plugins.kafka.consumer.KafkaCustomConsumer - Assigned partition test-encrypted-0
2024-02-15T09:38:29,901 [kafka-buffer-1] [buffer] INFO  org.opensearch.dataprepper.plugins.kafka.consumer.KafkaCustomConsumer - Partition test-encrypted-0 offsets: beginningOffset: 0, endOffset: 15, committedOffset: 15
2024-02-15T09:38:29,903 [kafka-buffer-1] [buffer] INFO  org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-data-prepper-1, groupId=data-prepper] Setting offset for partition test-encrypted-0 to the committed offset FetchPosition{offset=15, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 0 rack: null)], epoch=0}}

And here are some sample logs from outside of Kafka

2024-02-15T09:38:26,926 [main] [] INFO  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [kafka-buffer-pipeline] - Initiating pipeline execution
2024-02-15T09:38:26,927 [kafka-buffer-pipeline-sink-worker-2-thread-1] [] INFO  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [kafka-buffer-pipeline] Sink is ready, starting source...

Check List

• New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
  • New functionality has javadoc added
• Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[chenqi0805]

Shall we invoke MDC.clear() after the run?

[dlvenable]

I don't think it really matters because the thread will end. So the ThreadLocalContext is lost. But, I could do it.