in_mqtt: add test code and support payload_key

[nokute78]

This patch is to

• Add test code for in_mqtt plugin
• Support payload_key MQTT input plugin: store mqtt payload under own field in output record #7845

Key	Description
Payload_Key	Specify the key where the payload key/value will be preserved.

When MQTT payload is {"key1": 123, "key2": 456} and Payload_Key is payload, output will be

{"topic"=>"some/topic", "payload"=>{"key1"=>123, "key2"=>456}}

---

Enter [N/A] in the box, if an item is not applicable to your change.

Testing
Before we can approve your change; please submit the following in a comment:

• Example configuration file for the change
• Debug log output from testing the change

• Attached Valgrind output that shows no leaks or memory corruption was found

If this is a change to packaging of containers or native binaries then please confirm it works for all targets.

• [N/A] Run local packaging test showing all targets (including any new ones) build.
• [N/A] Set ok-package-test label to test for all targets (requires maintainer to do).

Documentation

• Documentation required for this feature

Backporting

• [N/A] Backport to latest stable release.

Configuration

[INPUT]
    name mqtt
    payload_key payload

[OUTPUT]
    name stdout
    match *

Valgrind output

Sent mqtt payload using

mosquitto_pub  -m '{"key1": 123, "key2": 456}' -t some/topic

$ valgrind --leak-check=full bin/fluent-bit -c a.conf 
==22052== Memcheck, a memory error detector
==22052== Copyright (C) 2002-2017, and GNU GPL'd, by Julian Seward et al.
==22052== Using Valgrind-3.18.1 and LibVEX; rerun with -h for copyright info
==22052== Command: bin/fluent-bit -c a.conf
==22052== 
Fluent Bit v2.1.9
* Copyright (C) 2015-2022 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2023/08/27 09:12:26] [ info] [fluent bit] version=2.1.9, commit=a2105c2c25, pid=22052
[2023/08/27 09:12:26] [ info] [storage] ver=1.2.0, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2023/08/27 09:12:26] [ info] [cmetrics] version=0.6.3
[2023/08/27 09:12:26] [ info] [ctraces ] version=0.3.1
[2023/08/27 09:12:26] [ info] [input:mqtt:mqtt.0] initializing
[2023/08/27 09:12:26] [ info] [input:mqtt:mqtt.0] storage_strategy='memory' (memory only)
[2023/08/27 09:12:27] [ info] [output:stdout:stdout.0] worker #0 started
[2023/08/27 09:12:27] [ info] [sp] stream processor started
[0] mqtt.0: [[1693095151.781405905, {}], {"topic"=>"some/topic", "payload"=>{"key1"=>123, "key2"=>456}}]
^C[2023/08/27 09:12:34] [engine] caught signal (SIGINT)
[2023/08/27 09:12:34] [ warn] [engine] service will shutdown in max 5 seconds
[2023/08/27 09:12:34] [ info] [engine] service has stopped (0 pending tasks)
[2023/08/27 09:12:34] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2023/08/27 09:12:34] [ info] [output:stdout:stdout.0] thread worker #0 stopped
==22052== 
==22052== HEAP SUMMARY:
==22052==     in use at exit: 0 bytes in 0 blocks
==22052==   total heap usage: 1,574 allocs, 1,574 frees, 827,772 bytes allocated
==22052== 
==22052== All heap blocks were freed -- no leaks are possible
==22052== 
==22052== For lists of detected and suppressed errors, rerun with: -s
==22052== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

---

Fluent Bit is licensed under Apache 2.0, by submitting this pull request I understand that this code will be released under the terms of that license.