fix(backend): fail fast on unreachable kafka brokers and add e2e cove… - #169
Open
harshitg927 wants to merge 1 commit into
Open
fix(backend): fail fast on unreachable kafka brokers and add e2e cove…#169harshitg927 wants to merge 1 commit into
harshitg927 wants to merge 1 commit into
Conversation
…rage Signed-off-by: Harshit Gandhi <gandhiharshit716@gmail.com>
📝 WalkthroughWalkthroughKafka startup now validates broker reachability with retries and cleanup on failure. New tests cover unreachable brokers, invalid SASL credentials, two-pod routing, and broker restarts. The change also adds Kafka configuration, Docker dependencies, and a local Redpanda startup script. ChangesKafka backend validation and integration
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
Suggested labels: Suggested reviewers: Sequence Diagram(s)sequenceDiagram
participant TwoPodKafkaTest
participant kafkaTestPod
participant KafkaBroker
participant recordingClientAdapter
TwoPodKafkaTest->>kafkaTestPod: create two connected pods
kafkaTestPod->>KafkaBroker: publish and consume synchronizer records
TwoPodKafkaTest->>KafkaBroker: publish cluster-specific command
KafkaBroker-->>kafkaTestPod: deliver record to matching consumer group
kafkaTestPod->>recordingClientAdapter: report received object
kafkaTestPod->>KafkaBroker: publish outbound cluster event
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
Author
|
@matthyx , PTAL |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
adapters/backend/v1/kafka_e2e_test.go (1)
73-81: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winLog the error from
client.Start/server.Startfor easier CI debugging.Both goroutines discard the return value of
Start. If the synchronizer client or server fails internally, the test only surfaces this later as a generic timeout (for example, at therequire.Eventuallyon line 83-85, or at a later channel wait), without the actual root cause. Logging the error witht.Loghelps diagnose flaky CI failures faster.♻️ Proposed improvement
- go func() { _ = client.Start(podCtx) }() - go func() { _ = server.Start(podCtx) }() + go func() { + if err := client.Start(podCtx); err != nil { + t.Logf("kafka test client stopped: %v", err) + } + }() + go func() { + if err := server.Start(podCtx); err != nil { + t.Logf("kafka test server stopped: %v", err) + } + }()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapters/backend/v1/kafka_e2e_test.go` around lines 73 - 81, Update the goroutines invoking client.Start and server.Start to capture their returned errors and log them with t.Log, while preserving the existing asynchronous startup behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@adapters/backend/v1/kafka_e2e_test.go`:
- Around line 73-81: Update the goroutines invoking client.Start and
server.Start to capture their returned errors and log them with t.Log, while
preserving the existing asynchronous startup behavior.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 9797f704-003f-40bf-bc7f-0e547c22c2f6
📒 Files selected for processing (8)
adapters/backend/v1/factory_test.goadapters/backend/v1/kafka.goadapters/backend/v1/kafka_e2e_test.goadapters/backend/v1/kafka_integration_test.goadapters/backend/v1/kafka_security_integration_test.goconfiguration/server-kafka/config.jsongo.modscripts/kafka.sh
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
…rage
Overview
Completes Phase 2 of the pluggable message queue proposal, closing the remaining gaps in the Kafka backend merged in #166 / #167.
Current behavior:
kgo.NewClientconnects lazily, so a synchronizer server configured with a wrong broker address, bad SASL credentials, or a failed TLS handshake starts successfully, serves/healthz200, and then silently never syncs. The Pulsar client connects eagerly and fails fast, so the Kafka path regressed this.New behavior: the Kafka factory pings both clients before returning, retrying on a budget that mirrors the Pulsar client (20 attempts at 3s). On failure the server exits with a fatal error instead of running healthy but dead.
Pingissues a Metadata request, so it covers TCP, TLS and SASL - note it does not validate that topics exist.Also adds the test coverage the proposal's Phase 2 testing strategy called for, plus a local Kafka development path:
TestKafkaMessageReader_SurvivesBrokerRestart) - the TC09 equivalent.TestKafkaE2E_TwoPodsRouteToConnectedClients) - two server instances, each with its own consumer group and a websocket-connected cluster; asserts every pod consumes the whole inbound topic but routes only what is addressed to a cluster connected to it, and that cluster-originated events reach.outwith the{account}/{cluster}partition key and expected headers.TestKafkaSecurity_BadCredentialsFailStartup).scripts/kafka.sh+configuration/server-kafka/config.jsonfor running the server against Kafka locally.Additional Information
A pre-existing data race is visible under
-race. The new E2E test exercisesAdapter.StartandAdapter.IsRelatedconcurrently, as the real server does when a cluster connects while the reader dispatches. That trips a bug ingoradd/maps, whereSafeMap.Load()readsm.itemsbefore takingRLockwhileSet()writes it under the write lock.This is not introduced here:
go test -race ./core/...already fails onmainwith several races in untouched code. CI is unaffected -pr-created.yamlsetsCGO_ENABLED: 0and the reusable workflow gates its-racestep onCGO_ENABLED == 1, so that step is skipped. Bumpinggoradd/mapsdoes not help (v1.3.0 has the identical unsynchronized read); the real fix is replacingSafeMapinadapter.go, which is out of scope here. Happy to open a separate issue.Why the restart test replaces the broker instead of restarting it. An in-place
Stop/Startof the testcontainers Redpanda module hangs forever:rpkrewrites/etc/redpanda/redpanda.yamlon first boot, stripping the# Injected by testcontainersmarker its entrypoint waits for. The container is therefore rebuilt on a pinned host port - the ephemeral port also changes across a restart, which would break the advertised listener and leave clients unable to reconnect.The test asserts recovery, not redelivery. The
.inreader is at-most-once by design (AtEnd, no commits, throwaway per-pod group), so messages produced during an outage are not expected to survive it.Why the E2E lives in the root module. The
tests/module embeds the realarmosec/event-ingester-service, which is Pulsar-only, so a Kafka E2E cannot live there until Phase 3. A plain franz-go consumer stands in for the ingester on.out.go.mod:docker/dockeranddocker/go-connectionsmove from indirect to direct - needed to pin the broker's host port in the restart test.Still out of scope: Phase 3 (ingester Kafka support) and Phase 4 (Helm + docs). The full
TC01–TC13matrix against Kafka remains blocked on Phase 3.How to Test
Manual end-to-end against a local broker:
To exercise the fail-fast path, point
bootstrapServersinconfiguration/server-kafka/config.jsonat a dead address and confirm the server logs retry warnings and then exits with a fatal error, rather than starting and serving/healthz.Related issues/PRs
Checklist before requesting a review
Summary by CodeRabbit