Skip to content

fix(kafka): retry mqtt producer startup - #1526

Merged
xxx7xxxx merged 2 commits into
easegress-io:mainfrom
xxx7xxxx:dev
Oct 13, 2026
Merged

xxx7xxxx merged 2 commits into
easegress-io:mainfrom
xxx7xxxx:dev

Conversation

@xxx7xxxx

Copy link
Copy Markdown
Contributor

Summary

Fix #1525

  • KafkaMQTT no longer fails pipeline initialization when Kafka is unavailable and retries producer creation until ready.
  • Document the startup retry behavior and refresh the filter reference URL example.

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This PR updates the KafkaMQTT filter to avoid failing pipeline initialization when Kafka is unavailable at startup by retrying sarama async producer creation in the background, while exposing a new producerUnavailable result and documenting the behavior.

Changes:

  • Make KafkaMQTT producer creation non-fatal on init failure, with exponential-backoff retry until ready.
  • Add runtime status reporting (Health/Ready) and a new filter result producerUnavailable returned while the producer is not available.
  • Update tests to cover retry-on-startup and close-before-reconnect behavior, and document the retry behavior in the MQTT proxy cookbook.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Show a summary per file
File Description
pkg/filters/kafka/spec.go Makes the mqtt/KV map optional in the filter spec.
pkg/filters/kafka/kafka.go Implements producer retry logic, producer-unavailable result, and status reporting.
pkg/filters/kafka/kafka_test.go Adds tests for startup retry, close behavior, and optional mqtt map spec validation.
docs/07.Reference/7.02.Filters.md Minor doc example/formatting adjustments (unrelated to KafkaMQTT behavior).
docs/03.Advanced-Cookbook/3.01.MQTT-Proxy.md Documents KafkaMQTT startup retry and producerUnavailable behavior.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Comment thread pkg/filters/kafka/kafka.go Outdated
Comment on lines +269 to +274
if k.producer != nil {
return &Status{
Health: "ready",
Ready: true,
}
}
@xxx7xxxx
xxx7xxxx added this pull request to the merge queue Oct 13, 2026
Merged via the queue into easegress-io:main with commit 48bacc2 Oct 13, 2026
10 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Bug]: The KafkaMQTT filter does not automatically reconnect to Kafka if the Kafka broker is unavailable when Easegress starts.

2 participants