kafka_out failures don't appear as failures in metrics

[dhumphries-sainsburys]

Bug Report

Describe the bug
When kafka is in an unavailable state it appears the kafka_out plugin drops events and doesn't report these as errors in the expected way. Specifically none of the error named metrics such as fluentbit_output_retries_total & fluentbit_output_dropped_records_total show any errors or dropped records despite there being many in the logs.

To Reproduce
Configure fluentbit to tail logs from eks as standaard
Configure fluentbit to send to kafka
Config like below

[SERVICE]
    Flush 1
    Daemon Off
    Log_Level info
    Parsers_File parsers.conf
    Parsers_File custom_parsers.conf
    HTTP_Server On
    HTTP_Listen 0.0.0.0
    HTTP_Port 2020

[INPUT]
    Name tail
    Path /var/log/containers/*.log
    Tag kube.*
    Parser cri-custom
    Mem_Buf_Limit 10MB
    Skip_Long_Lines On
    Read_from_Head True
    DB /var/log/fluent-bit-tail-db

[FILTER]
    Name kubernetes
    Match kube.var.log.containers.*
    Merge_Log On
    Merge_Log_Key applog
    Keep_Log Off
    K8S-Logging.Parser On
    K8S-Logging.Exclude On
    Annotations Off

[FILTER]
    Name rewrite_tag
    Match kube.var.log.containers.*
    Rule $kubernetes['namespace_name'] .* kube.$kubernetes['namespace_name'] false

[FILTER]
    Name parser
    Match *
    Key_Name applog_message
    Parser mpe-parser
    Reserve_Data On
    Preserve_Key On

[OUTPUT]
    Name kafka
    Log_Level error
    Match *
    Brokers kafka-logs-kafka-internal-bootstrap:9092
    Topics turtle-logs-os
    timestamp_format iso8601
    rdkafka.security.protocol sasl_ssl
    rdkafka.enable.ssl.certificate.verification false
    rdkafka.sasl.mechanism SCRAM-SHA-512
    rdkafka.sasl.username user-writer
    rdkafka.sasl.password ${KAFKA_USER_WRITER}

[OUTPUT]
    name prometheus_exporter
    match  *
    host 0.0.0.0
    port  2021

• Steps to reproduce the problem:
  Disable kafka either by turning it off or breaking authentication to it

Expected behavior

• Ideally error rate reported and no events dropped at all

Your Environment

• Version used: 3.0.4
• Configuration: As above
• Environment name and version (e.g. Kubernetes? What version?): EKS 1.30
• Server type and version: Bottlerocket 1.30
• Operating System and version: Bottlerocket 1.30
• Filters and plugins: No additional

Additional context

Output to kafka can fail silently and doesn't seem possible to monitor