Fix coordinator broker identity mismatch

[kamir]

Summary

Problem

• FindCoordinator could return a broker ID that was missing from Metadata, causing Kafka clients to retry because the coordinator identity could not be reconciled with broker metadata.

Solution

• Select the coordinator broker from the metadata snapshot first, falling back to local broker info only if metadata is unavailable, and default broker ID to 0 to align with operator snapshots.

How It Works

• On FindCoordinator, the handler loads metadata and returns a broker ID that exists in the advertised broker list; if the local ID is missing, it returns the first metadata broker, and if metadata fails, it uses local broker info.

Why Tests/Benchmarks Missed It

• E2E Franz runs use in-memory metadata seeded from the same broker info, operator e2e paths use broker IDs derived from pod ordinals that match snapshots, and benchmarks use kcat produce/consume without consumer groups (no FindCoordinator).

Testing

• go test ./...
• make test-produce-consume (broker changes)
• make test-consumer-group (group changes)

Checklist

• Added/updated unit tests for new logic
• Added/updated e2e coverage for bug fixes
• Added license headers to new files

[novatechflow]

.github/pull_request_template.md is used to prefill PR's. This file has not to be changed. Please revert to the original.

[novatechflow]

this PR will fix issue #70

[novatechflow]

thank you @kamir - great catch!