Processor nats_kv removes all previously set meta variables on the operations get and get_revision

I have some issues when using the nats_kv processor. Whenever I have metadata and use either get or get_revision all previous meta is gone.

From trying to debug the issue I found that on get and get_revision then function newMessageFromKVEntry is called to create a new message (without any previous meta).
For the other operations a msg.Copy() is used and metadata are added.

I propose a new function that copies existing meta as well....

func newMessageWithMetaFromKVEntry(entry jetstream.KeyValueEntry, src *service.Message) *service.Message {
	msg := service.NewMessage(entry.Value())
	//nolint:errcheck
	src.MetaWalkMut(func(key string, value any) error {
		msg.MetaSetMut(key, value)
		return nil
	})
	msg.MetaSetMut(metaKVKey, entry.Key())
	msg.MetaSetMut(metaKVBucket, entry.Bucket())
	msg.MetaSetMut(metaKVRevision, entry.Revision())
	msg.MetaSetMut(metaKVDelta, entry.Delta())
	msg.MetaSetMut(metaKVOperation, entry.Operation().String())
	msg.MetaSetMut(metaKVCreated, entry.Created())
	return msg
}

I can quite easily create a PR with this but there might be a better way to do it.

To recreate...

config:

input:
  stdin: {}

pipeline:
  processors:
    - mapping: |
          root.doc = this
          meta something = "blob"
    
    - nats_kv:
        urls: ["nats://localhost:4222"]
        bucket: test
        operation: put
        key: "blob"

    - log:
        level: INFO
        message: 'meta after put: ${! meta("something") }'

    - nats_kv:
        urls: ["nats://localhost:4222"]
        bucket: test
        operation: get
        key: "blob"

    - log:
        level: INFO
        message: 'meta after get: ${! meta("something") }'

output:
  stdout: {}

output:

INFO Running main config from specified file       @service=benthos benthos_version=v4.45.0-rc1-80-gd32a89f95 path=tmp/test.yml
INFO Listening for HTTP requests at: http://0.0.0.0:4195  @service=benthos
INFO Input type stdin is now active                @service=benthos label="" path=root.input
INFO Launching a Redpanda Connect instance, use CTRL+C to close  @service=benthos
INFO Output type stdout is now active              @service=benthos label="" path=root.output
"asdf"
INFO meta after put: blob                          @service=benthos custom_source=true label="" path=root.pipeline.processors.2
INFO meta after get: null                          @service=benthos custom_source=true label="" path=root.pipeline.processors.4
{"doc":"asdf"}

Start by locating the nats_kv processor paths for get and get_revision and the newMessageFromKVEntry function mentioned in the issue. Compare them with the other operations that use msg.Copy(), then reproduce the supplied configuration. Done means metadata set before the key-value lookup, including something, remains available after get and get_revision while the key-value metadata is still present.