Kafka sink metrics incorrect after reload Kafka config.

[fcfangcc]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

After change the config of Kafka sink and reloading it(run vector with --watch-config), Kafka's metrics incorrect. Kafka metrics is emit withKafkaStatisticsContext, after reload config and new producer created. KafkaStatisticsReceived.statistics.txmsg_bytes will from with 0. At this time, Counter metrics(Like kafka_produced_messages_bytes_total) the value of the indicator will not increase until the size is greater than the previous size. https://github.com/vectordotdev/vector/blob/21c3e68bfd2406e703b6587ed474e4e08a3c9c0f/lib/vector-core/src/metrics/mod.rs#L195-L201

Configuration

sources:
  dummy_logs:
    type: "demo_logs"
    format: "syslog"
    interval: 1

  internal_metrics:
    type: internal_metrics

transforms:
  parse_logs:
    type: "remap"
    inputs: ["dummy_logs"]
    source: |
      . = parse_syslog!(string!(.message))

sinks:
  log_to_kafka:
    type: kafka
    inputs:
      - parse_logs
    bootstrap_servers: xxx:9092
    topic: xxx
    encoding:
      codec: json
    compression: gzip

  prometheus_exporter:
    inputs:
      - "*_metrics"
    type: prometheus_exporter

Version

0.39.0

Debug Output

No response

Example Data

No response

Additional Context

No response

References

No response