[kafka] Handle configuration errors

[khushijain21]

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

[github-actions]

🤖 GitHub comments

Expand to view the GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @khushijain21? 🙏.
For such, you'll need to label your PR with:

• The upcoming major version of the Elastic Stack
• The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit
• backport-active-all is the label that automatically backports to all active branches.
• backport-active-8 is the label that automatically backports to all active minor branches for the 8 major.
• backport-active-9 is the label that automatically backports to all active minor branches for the 9 major.

[elasticmachine]

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

[belimawr]

I don't think dropping messages on all configuration errors is the correct approach here.

The goal is to fix #44619 by bringing back the behaviour of dropping messages if they're too large. Other configuration issues should be handled the same way they're handled today.

The problem with dropping messages on any configuration errors is that a configuration error can be fixed with user intervention. In that case we do not want inputs that keep track of their progress (like Filestream) to ingest all available data just to have it discarded by the output. Effectively creating another instance of #43250.

Even though it's not ideal, I prefer to compare the error string looking for Attempt to produce message larger than configured Producer.MaxMessageBytes (see the PR introducing the change), if it matches, we then drop the event like we used to. Any other configuration error we still retry.

[faec]

Looks mostly fine, a couple small requests. (I'm about to be on PTO for a few days so feel free to merge in my absence once other reviewers are happy.)

[khushijain21]

@tiago can you take a look again. I have ensured we don't drop all configuration related and errors and also incorporated Fae's remarks

[belimawr]

The code looks good now!

Sorry for not mentioning that before, but could you add some tests for the code you added? Especially now that we're looking at the error string we should have a test enduring the behaviour is not going to change.

[belimawr]

It looks great now!

I added some minor comments about formatting, feel free to ignore them ;)

[github-actions]

@Mergifyio backport 8.17 8.18 8.19 9.0 9.1

[mergify]

backport 8.17 8.18 8.19 9.0 9.1

✅ Backports have been created

Details

• #45279 [8.17](backport #45128) [kafka] Handle configuration errors has been created for branch 8.17 but encountered conflicts
• #45280 [8.18](backport #45128) [kafka] Handle configuration errors has been created for branch 8.18 but encountered conflicts
• #45281 [8.19](backport #45128) [kafka] Handle configuration errors has been created for branch 8.19
• #45282 [9.0](backport #45128) [kafka] Handle configuration errors has been created for branch 9.0 but encountered conflicts
• #45283 [9.1](backport #45128) [kafka] Handle configuration errors has been created for branch 9.1