[8.19](backport #45128) [kafka] Handle configuration errors

[mergify]

Proposed commit message

This PR handles kafka configuration errors by dropping the message. Since configuration errors are of permanent error type and should not be retried.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

How to test this PR locally

Bring kafka server up using following command

cd filebeat
mage docker:composeUp

Use following filebeat.yml

filebeat.inputs:
  - type: filestream
    id: filestream-input-id
    enabled: true
    paths:
      - ./test.log
    file_identity.native: ~
    prospector.scanner.fingerprint.enabled: false      

output:
  kafka:
    hosts: ["localhost:9092", "localhost:9094"]
    topic: beats
    max_message_bytes: 10

and you can see this error

{
  "log.level": "error",
  "@timestamp": "2025-07-02T15:28:04.373+0530",
  "log.logger": "kafka",
  "log.origin": {
    "function": "github.com/elastic/beats/v7/libbeat/outputs/kafka.(*msgRef).fail",
    "file.name": "kafka/client.go",
    "file.line": 396
  },
  "message": "Kafka (topic=beats): dropping message due to invalid configuration kafka: invalid configuration (Attempt to produce message larger than configured Producer.MaxMessageBytes: 851 > 10)",
  "service.name": "filebeat",
  "ecs.version": "1.6.0"
}

Related issues

• Closes Kafka output retries 'too large messages' (message len > max_message_bytes) instead of dropping it (>= v8.18.0) #44619

---

This is an automatic backport of pull request #45128 done by [Mergify](https://mergify.com).

[elasticmachine]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)