feat(backend): add Kafka message queue backend with franz-go - #166
Conversation
📝 WalkthroughWalkthroughAdds configurable Kafka support alongside Pulsar, including franz-go producer and reader implementations, Redpanda integration tests, dependency updates, and graceful HTTP server shutdown. ChangesKafka backend
Server lifecycle
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant KafkaMessageReader
participant Kafka
participant messaging.MessageHandler
participant adapters.Adapter
KafkaMessageReader->>Kafka: Poll records
Kafka-->>KafkaMessageReader: Return fetched records
KafkaMessageReader->>messaging.MessageHandler: Dispatch IncomingMessage
messaging.MessageHandler->>adapters.Adapter: Handle message callback
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 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 |
|
@matthyx One scoping question on SASL/TLS. SASL/TLS is required for any real deployment, so I want to get it right. In this PR I've laid the groundwork but deferred the enforcement: the kafkaConfig block already parses all the security fields (securityProtocol, saslMechanism, saslUsername/saslPassword, tlsEnabled, tlsCaCertPath), and anything other than PLAINTEXT currently fails loudly at startup rather than being silently ignored. So wiring the actual SASL/SCRAM + TLS dialer into the franz-go client is the only remaining step - franz-go supports it out of the box. I kept it out of this PR to keep the core produce/consume + partition-key + 64 MB + fan-out changes focused and reviewable, with the integration tests running against a plaintext broker. Would you prefer to, add SASL/TLS in a follow-up PR (this one lands as PLAINTEXT-only, security enforcement + a secured-broker integration test come next), or fold it into this PR so v1 ships complete? |
|
you can add that as a follow-up 👍 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (2)
adapters/backend/v1/factory.go (1)
19-28: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winExplicit
type: "pulsar"is rejected instead of falling back.Only an absent
messageQueue.typefalls back to Pulsar; an explicit"pulsar"value hits thedefaultbranch and errors as "unknown message queue type". Since"kafka"is a documented valid value, operators are likely to try"pulsar"as its counterpart.♻️ Suggested fix
switch mq.Type { case "kafka": return newKafkaFromConfig(cfg) + case "pulsar": + return newPulsarFromConfig(cfg) default: return nil, fmt.Errorf("unknown message queue type %q", mq.Type) }🤖 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/factory.go` around lines 19 - 28, Update the message queue type switch in the backend factory to explicitly accept `"pulsar"` and return newPulsarFromConfig(cfg), while preserving the existing Kafka handling and unknown-type error behavior. Ensure both an omitted type and an explicit `"pulsar"` type select Pulsar.adapters/backend/v1/kafka_integration_test.go (1)
260-262: 🩺 Stability & Availability | 🔵 Trivial | ⚖️ Poor tradeoffFixed-sleep readiness waits are flake-prone but low priority for integration tests.
Both
TestKafkaMessageReader_DispatchesToAdapterandTestKafkaMessageReader_FanOutAcrossGroupsuse a hardcodedtime.Sleep(5 * time.Second)to wait for the reader(s) to join their consumer group before producing. On a slow CI runner this could occasionally under-wait and fail. Not urgent given the 60s downstreamselecttimeout provides some cushion, but a readiness poll (e.g., checking group members viakadm) would be more robust.Also applies to: 308-309
🤖 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_integration_test.go` around lines 260 - 262, Replace the fixed 5-second sleeps in TestKafkaMessageReader_DispatchesToAdapter and TestKafkaMessageReader_FanOutAcrossGroups with readiness polling that verifies the reader consumer group members through the existing Kafka admin mechanism before producing messages. Wait until all expected readers have joined, while retaining a bounded timeout and the existing downstream message assertions.
🤖 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.
Inline comments:
In `@adapters/backend/v1/kafka_integration_test.go`:
- Around line 343-354: Update pollRecords to explicitly detect pollCtx
expiration after PollFetches returns without completing the requested record
count, and fail the test with a clear timeout message instead of continuing the
loop. Preserve existing fetch-error handling and successful accumulation
behavior.
In `@adapters/backend/v1/kafka.go`:
- Around line 38-49: Update newKafkaFromConfig to validate
kafkaCfg.ProducerTopic and kafkaCfg.ConsumerTopic after BootstrapServers
validation, returning clear errors when either is empty before constructing
messaging components. Preserve the existing securityProtocol validation and
ensure both topics are required so producer and consumer paths receive valid
defaults.
- Around line 279-353: Move ownership of closing messageChannel to readerLoop,
ensuring it closes only after the fetch loop and any in-progress batch delivery
finish, and remove the worker-supervisor close after wg.Wait. Keep
listenOnMessageChannel’s !ok handling so workers exit when the writer closes the
channel, while retaining context cancellation as a secondary exit path.
In `@config/config.go`:
- Around line 50-51: Correct the flow-direction comments for ProducerTopic and
ConsumerTopic in the configuration definitions: ProducerTopic is used by
NewKafkaMessageProducer for backend → cluster (.out), while ConsumerTopic is
used by NewKafkaMessageReader for cluster → backend (.in). Leave the field names
and mapstructure tags unchanged.
---
Nitpick comments:
In `@adapters/backend/v1/factory.go`:
- Around line 19-28: Update the message queue type switch in the backend factory
to explicitly accept `"pulsar"` and return newPulsarFromConfig(cfg), while
preserving the existing Kafka handling and unknown-type error behavior. Ensure
both an omitted type and an explicit `"pulsar"` type select Pulsar.
In `@adapters/backend/v1/kafka_integration_test.go`:
- Around line 260-262: Replace the fixed 5-second sleeps in
TestKafkaMessageReader_DispatchesToAdapter and
TestKafkaMessageReader_FanOutAcrossGroups with readiness polling that verifies
the reader consumer group members through the existing Kafka admin mechanism
before producing messages. Wait until all expected readers have joined, while
retaining a bounded timeout and the existing downstream message assertions.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 655667a3-f83f-4aeb-867f-03a672ab9396
⛔ Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
📒 Files selected for processing (5)
adapters/backend/v1/factory.goadapters/backend/v1/kafka.goadapters/backend/v1/kafka_integration_test.goconfig/config.gogo.mod
| func pollRecords(t *testing.T, ctx context.Context, consumer *kgo.Client, n int) []*kgo.Record { | ||
| t.Helper() | ||
| pollCtx, cancel := context.WithTimeout(ctx, 30*time.Second) | ||
| defer cancel() | ||
| records := make([]*kgo.Record, 0, n) | ||
| for len(records) < n { | ||
| fetches := consumer.PollFetches(pollCtx) | ||
| require.NoError(t, fetches.Err()) | ||
| records = append(records, fetches.Records()...) | ||
| } | ||
| return records | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
pollRecords can busy-loop with an unhelpful failure once pollCtx expires.
The 30s pollCtx is created once before the loop; once it expires, PollFetches(pollCtx) should return immediately, but if it does so without records or an error, for len(records) < n keeps looping tightly rather than failing with a clear message, until the outer test-suite timeout eventually intervenes.
✅ Suggested fix
for len(records) < n {
fetches := consumer.PollFetches(pollCtx)
require.NoError(t, fetches.Err())
records = append(records, fetches.Records()...)
+ if pollCtx.Err() != nil && len(records) < n {
+ t.Fatalf("timed out waiting for %d records, got %d", n, len(records))
+ }
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func pollRecords(t *testing.T, ctx context.Context, consumer *kgo.Client, n int) []*kgo.Record { | |
| t.Helper() | |
| pollCtx, cancel := context.WithTimeout(ctx, 30*time.Second) | |
| defer cancel() | |
| records := make([]*kgo.Record, 0, n) | |
| for len(records) < n { | |
| fetches := consumer.PollFetches(pollCtx) | |
| require.NoError(t, fetches.Err()) | |
| records = append(records, fetches.Records()...) | |
| } | |
| return records | |
| } | |
| func pollRecords(t *testing.T, ctx context.Context, consumer *kgo.Client, n int) []*kgo.Record { | |
| t.Helper() | |
| pollCtx, cancel := context.WithTimeout(ctx, 30*time.Second) | |
| defer cancel() | |
| records := make([]*kgo.Record, 0, n) | |
| for len(records) < n { | |
| fetches := consumer.PollFetches(pollCtx) | |
| require.NoError(t, fetches.Err()) | |
| records = append(records, fetches.Records()...) | |
| if pollCtx.Err() != nil && len(records) < n { | |
| t.Fatalf("timed out waiting for %d records, got %d", n, len(records)) | |
| } | |
| } | |
| return records | |
| } |
🤖 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_integration_test.go` around lines 343 - 354, Update
pollRecords to explicitly detect pollCtx expiration after PollFetches returns
without completing the requested record count, and fail the test with a clear
timeout message instead of continuing the loop. Preserve existing fetch-error
handling and successful accumulation behavior.
There was a problem hiding this comment.
I don't think this is an issue in practice. franz-go injects a fake fetch with ctx.Err() when the poll context is canceled (it's even mentioned in the PollRecords docstring: "If the context is canceled, a fake fetch will be injected with ctx.Err().").
So once the 30s deadline is reached, fetches.Err() returns context.DeadlineExceeded, and the require.NoError(t, fetches.Err()) on the very next line fails the test immediately. It never ends up busy-looping or waiting for the outer suite timeout.
Also, I think the suggested guard would never actually be hit, since require.NoError fails first. I'd prefer to leave this as-is, though I'm happy to replace the assertion with a more descriptive failure message if that's clearer than just context deadline exceeded.
@matthyx , what do you think?
|
Sure @matthyx , I will address the coderabbit's review first |
There was a problem hiding this comment.
🧹 Nitpick comments (2)
adapters/backend/v1/factory_test.go (2)
36-43: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAssert the specific error message to avoid false positives.
Using
require.Errorconfirms that an error occurred, but it might pass for the wrong reason (e.g., an unexpected panic or a different validation error). Consider asserting the exact error string to make the test more robust.♻️ Proposed refactor
func TestNewFromConfig_UnknownType(t *testing.T) { _, err := newFromConfig(config.Config{ Backend: config.Backend{ MessageQueue: &config.MessageQueueConfig{Type: "rabbitmq"}, }, }) - require.Error(t, err) + require.EqualError(t, err, `unknown message queue type "rabbitmq"`) }🤖 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/factory_test.go` around lines 36 - 43, Strengthen TestNewFromConfig_UnknownType by asserting the returned error’s exact message, not only that an error exists. Preserve the rabbitmq configuration and verify the message corresponds specifically to the unsupported backend type.
45-65: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider verifying the specific validation error messages.
Similar to the unknown type test, table-driven validation tests can yield false positives if they fail due to an unrelated error rather than the specific validation logic being tested. Adding an expected error string to the test cases and using
require.ErrorContainsensures each case fails for the correct reason.♻️ Proposed refactor (example with placeholder error messages)
func TestNewFromConfig_KafkaValidation(t *testing.T) { tests := []struct { - name string - kafka *config.KafkaConfig + name string + kafka *config.KafkaConfig + errContains string }{ - {name: "missing kafkaConfig", kafka: nil}, - {name: "missing bootstrapServers", kafka: &config.KafkaConfig{ProducerTopic: "out", ConsumerTopic: "in"}}, - {name: "missing producerTopic", kafka: &config.KafkaConfig{BootstrapServers: []string{"localhost:9092"}, ConsumerTopic: "in"}}, - {name: "missing consumerTopic", kafka: &config.KafkaConfig{BootstrapServers: []string{"localhost:9092"}, ProducerTopic: "out"}}, + {name: "missing kafkaConfig", kafka: nil, errContains: "kafka config is required"}, + {name: "missing bootstrapServers", kafka: &config.KafkaConfig{ProducerTopic: "out", ConsumerTopic: "in"}, errContains: "bootstrapServers"}, + {name: "missing producerTopic", kafka: &config.KafkaConfig{BootstrapServers: []string{"localhost:9092"}, ConsumerTopic: "in"}, errContains: "producerTopic"}, + {name: "missing consumerTopic", kafka: &config.KafkaConfig{BootstrapServers: []string{"localhost:9092"}, ProducerTopic: "out"}, errContains: "consumerTopic"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { _, err := newFromConfig(config.Config{ Backend: config.Backend{ MessageQueue: &config.MessageQueueConfig{Type: "kafka", KafkaConfig: tt.kafka}, }, }) - require.Error(t, err) + require.ErrorContains(t, err, tt.errContains) }) } }🤖 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/factory_test.go` around lines 45 - 65, Update TestNewFromConfig_KafkaValidation to include the expected validation message for each table case, then replace require.Error with require.ErrorContains using that expected message. Keep the existing cases and newFromConfig invocation unchanged so each test verifies the specific Kafka validation branch rather than merely any error.
🤖 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/factory_test.go`:
- Around line 36-43: Strengthen TestNewFromConfig_UnknownType by asserting the
returned error’s exact message, not only that an error exists. Preserve the
rabbitmq configuration and verify the message corresponds specifically to the
unsupported backend type.
- Around line 45-65: Update TestNewFromConfig_KafkaValidation to include the
expected validation message for each table case, then replace require.Error with
require.ErrorContains using that expected message. Keep the existing cases and
newFromConfig invocation unchanged so each test verifies the specific Kafka
validation branch rather than merely any error.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: dd56e6b1-1780-41b2-975f-c7e64b58c68e
📒 Files selected for processing (5)
adapters/backend/v1/factory.goadapters/backend/v1/factory_test.goadapters/backend/v1/kafka.goadapters/backend/v1/kafka_integration_test.goconfig/config.go
🚧 Files skipped from review as they are similar to previous changes (4)
- adapters/backend/v1/factory.go
- config/config.go
- adapters/backend/v1/kafka_integration_test.go
- adapters/backend/v1/kafka.go
matthyx
left a comment
matthyx
left a comment
There was a problem hiding this comment.
Review — Kafka backend (Phase 2)
Solid, well-scoped PR. The messaging-abstraction seam is respected, Pulsar stays the untouched default, and the async/worker patterns mirror the existing Pulsar reader. I verified the franz-go behaviors the design leans on (keyed sticky partitioner + default idempotent producer preserve per-partition order on .out; consumer-group ConsumeResetOffset(AtEnd) with auto-commit disabled gives the intended per-pod at-most-once fan-out on .in). No hard functional blockers — the earlier bot findings (topic validation, send-on-closed-channel ownership, explicit type: "pulsar") are already handled in the current head.
One item I'd fix before merge, plus two minor notes.
🔴 Should-fix (security): TLS/SASL config is silently ignored, not "failed loudly"
newKafkaFromConfig only guards SecurityProtocol (kafka.go:137-140). The other security fields — TLSEnabled, TLSCaCertPath, SASLMechanism, SASLUsername, SASLPassword — are parsed into KafkaConfig but never passed to kgo.NewClient, and are not validated. So a config like:
securityProtocol: "" # or "PLAINTEXT"
saslMechanism: "SCRAM-SHA-256"
saslUsername: "svc"
saslPassword: "..."
tlsEnabled: truepasses validation and connects in plaintext with no auth, silently. An operator who sets credentials but forgets/omits securityProtocol believes the connection is secured when it is not — which is exactly the "silently ignoring security config" failure the securityProtocol guard was added to prevent. Suggest rejecting up front if any SASL/TLS field is populated while enforcement is unimplemented, e.g.:
if kafkaCfg.TLSEnabled || kafkaCfg.SASLMechanism != "" || kafkaCfg.SASLUsername != "" {
return nil, fmt.Errorf("kafkaConfig TLS/SASL settings are not yet enforced (only PLAINTEXT); SASL/TLS ships in a follow-up")
}🟡 Minor: buffered .out records can be dropped on shutdown
KafkaMessageProducer.Close() calls client.Close() with no preceding Flush(). Per franz-go, Close() does not flush — buffered records are dropped. Unlike the Pulsar producer (DisableBatching: true, so little is buffered), this client batches (ProducerBatchCompression + 64 MB ProducerBatchMaxBytes), so more in-flight .out (cluster→backend) data is at risk. Compounding it: cmd/server/main.go has no signal handling and http.ListenAndServe blocks, so defer mq.Close() effectively never runs on SIGTERM — the process is killed with records still buffered. A p.client.Flush(ctx) before Close() (and, longer term, graceful shutdown on SIGTERM) would make the .out path — which is not at-most-once by design — actually reliable. Pre-existing for the shutdown wiring, but the batching change makes it more visible here.
🟡 Minor: ProducerBatchMaxBytes(int32(maxMessageBytes)) is unclamped
kafkaBrokerByteLimit clamps the broker limit to [100 MB, 1 GB], but the batch-max value is a bare int32(maxMessageBytes). A maxMessageBytes > 2 GiB config overflows int32 to a negative value (broker limit stays clamped at 1 GB, so they also diverge). Unrealistic today, but a small min() guard would make the sizing math robust.
Nice work overall — the design rationale in the comments made the ordering/fan-out semantics easy to verify.
|
@matthyx thanks, all three fixed. TLS/SASL - you're right, guarding only securityProtocol left the exact hole it was meant to close. Now rejects any populated TLS/SASL field too, with tests per field. Flush - confirmed in the franz-go source that Close() fails buffered records rather than flushing, so added a bounded Flush() first. Also wired up SIGTERM handling in main.go, since without it the deferred Close() never runs and the flush does nothing. That part touches the shared serving path (helps Pulsar too) - happy to split it out if you'd rather keep this PR Kafka-only. int32 clamp - done, and applied to FetchMaxBytes as well since it had the same problem. Build, vet, and the Kafka + Pulsar suites all pass. I will open a follow-up PR with SASL/TLS enforcement when this one is merged. |
Signed-off-by: Harshit Gandhi <gandhiharshit716@gmail.com>
Signed-off-by: Harshit Gandhi <gandhiharshit716@gmail.com>
…utdown Signed-off-by: Harshit Gandhi <gandhiharshit716@gmail.com>
56390f1 to
3af32ab
Compare
matthyx
left a comment
matthyx
left a comment
There was a problem hiding this comment.
Re-review — all three points addressed ✅
Verified the follow-up commit (fix(backend): reject unenforced kafka TLS/SASL config and flush on shutdown):
- 🔴 Security (was the blocking one):
newKafkaFromConfignow rejects any populatedTLSEnabled/TLSCaCertPath/SASLMechanism/SASLUsername/SASLPasswordwith a clear "not yet enforced" error, alongside the existingsecurityProtocolguard — no more silent plaintext. Covered by the six new table cases infactory_test.go. Resolved. - 🟡 Flush on shutdown:
KafkaMessageProducer.Close()nowFlush()es under a bounded 30s context beforeclient.Close(), logging on failure. Correct pattern for franz-go. Resolved. - 🟡 int32 overflow: new
kafkaRecordByteLimitclamps tokafkaMaxBrokerBytes(1 GiB) before the int32 cast, and it's applied consistently to bothProducerBatchMaxBytesand the reader'sFetchMaxBytes. Resolved.
No further blockers from my side. LGTM. (The graceful-shutdown / SIGTERM wiring in cmd/server/main.go remains a separate, pre-existing item — out of scope for this PR.)
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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.
Inline comments:
In `@cmd/server/main.go`:
- Around line 124-140: Synchronize the signal-handling goroutine with main so
the process waits for srv.Shutdown to return before exiting and running deferred
cleanup; add a completion channel around the existing shutdown goroutine and
await it after ListenAndServe returns. In the ListenAndServe error path, return
immediately for real errors instead of waiting for a signal that cannot arrive,
while preserving normal shutdown handling for http.ErrServerClosed.
- Around line 85-87: Update the http.Server initialization in main to set a
finite ReadHeaderTimeout appropriate for the internet-facing service, while
preserving the existing Addr and Handler configuration.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: f13924df-ff34-4888-aab0-087fe9185199
⛔ Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
📒 Files selected for processing (7)
adapters/backend/v1/factory.goadapters/backend/v1/factory_test.goadapters/backend/v1/kafka.goadapters/backend/v1/kafka_integration_test.gocmd/server/main.goconfig/config.gogo.mod
🚧 Files skipped from review as they are similar to previous changes (5)
- adapters/backend/v1/factory_test.go
- config/config.go
- adapters/backend/v1/factory.go
- go.mod
- adapters/backend/v1/kafka_integration_test.go
matthyx
left a comment
matthyx
left a comment
There was a problem hiding this comment.
Triage of the bot findings on the new main.go shutdown path
I re-verified the outstanding automated comments. One is a real blocker, one is worth doing with a caveat, and one earlier finding was a false positive that was correctly rejected.
🔴 Confirmed blocker — main.go: nothing waits for srv.Shutdown to return
The Critical finding on the shutdown race is correct, and it matters more here than the generic version of this advice, because it defeats the Flush()-on-close fix added in this same commit:
- Signal arrives → the goroutine calls
srv.Shutdown(shutdownCtx). Shutdownmakessrv.ListenAndServe()inmainreturnhttp.ErrServerClosedimmediately.mainfalls out of theif, returns, and runsdefer mq.Close()/defer cancel().- Process exits — while the shutdown goroutine is still draining, and while websocket handlers may still be producing records.
So mq.Close() can flush while connections are still live and producing, and the 30s shutdownTimeout is never actually honoured. This is still an improvement over the old _ = http.ListenAndServe(...) (the deferred close now runs at all), but the drain is cut short. Needs a done-channel so main blocks until Shutdown returns — and, as the bot notes, an early return if ListenAndServe fails with a real startup error (e.g. port in use), otherwise you deadlock waiting for a signal that never comes.
Ordering worth getting right while you're in there: drain HTTP → then mq.Close() (flush). Right now the defer ordering makes that racy rather than sequential.
🟡 ReadHeaderTimeout — worth adding, but only that one
Reasonable hardening for a listener clusters dial into from outside. One caveat, since it's easy to over-apply: add ReadHeaderTimeout only — do not add ReadTimeout or WriteTimeout to this server. Those bound the whole request lifetime and would tear down the long-lived websocket connections this service exists to hold open. ReadHeaderTimeout is the websocket-safe one because it bounds header reads only.
✅ Correctly rejected — the config.go topic-direction comment
For the record, the earlier suggestion to flip the ProducerTopic/ConsumerTopic flow comments was wrong, and not applying it was the right call. The synchronizer server sits between the clusters and the backend, so when it produces it is forwarding cluster→backend (.out); when it reads it is receiving backend→cluster (.in). The suggested diff also contradicted the design proposal by relabelling .in as cluster→backend. The current wording ("server produces here…", "server consumes here…") removes the ambiguity that tripped the bot without inverting the meaning — good resolution.
The pollRecords nit was also fairly rebutted: franz-go injects a fake fetch carrying ctx.Err() on poll-context cancellation, so require.NoError(t, fetches.Err()) fails the test with a clear error rather than busy-looping.
…aderTimeout Signed-off-by: Harshit Gandhi <gandhiharshit716@gmail.com>
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
cmd/server/main.go (1)
129-150: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy liftTrack WebSocket sessions during shutdown.
srv.Shutdowndoes not wait for hijacked connections, so the detached WebSocket goroutines can keep using the adapter/message queue aftershutdownDoneunblocks. Track active sessions and close/cancel them before waiting for shutdown.🤖 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 `@cmd/server/main.go` around lines 129 - 150, Update the shutdown flow around srv.Shutdown and shutdownDone to track active WebSocket sessions, then close or cancel every tracked session before shutdown completion is signaled. Ensure detached WebSocket goroutines have stopped using the adapter/message queue before the existing `<-shutdownDone` wait returns, integrating the tracking with the session lifecycle rather than only HTTP server shutdown.
🤖 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.
Outside diff comments:
In `@cmd/server/main.go`:
- Around line 129-150: Update the shutdown flow around srv.Shutdown and
shutdownDone to track active WebSocket sessions, then close or cancel every
tracked session before shutdown completion is signaled. Ensure detached
WebSocket goroutines have stopped using the adapter/message queue before the
existing `<-shutdownDone` wait returns, integrating the tracking with the
session lifecycle rather than only HTTP server shutdown.
matthyx
left a comment
matthyx
left a comment
There was a problem hiding this comment.
Re-review — shutdown race resolved ✅
Verified fix(server): block on graceful shutdown before cleanup and add ReadHeaderTimeout by tracing both exit paths and building the branch locally.
Shutdown race — fixed correctly. Both paths check out:
- Signal path: signal →
srv.Shutdown()→ListenAndServe()returnsErrServerClosed→ theifdoes not fire, somainfalls through to<-shutdownDoneand blocks until the goroutine'sShutdownactually returns. The 30sshutdownTimeoutis now genuinely honoured. - Startup-failure path: a real
ListenAndServeerror logs andreturns before<-shutdownDone, so there's no deadlock waiting on a signal that never arrives. That was the easy trap here and it's handled.
Cleanup ordering is right too — worth calling out since it's what the whole thread was about. defer cancel() is registered before defer mq.Close(), so LIFO gives: drain HTTP → mq.Close() (flush buffered producer records) → cancel(). The flush now happens after connections are drained and before the reader context is torn down, which is the sequence that makes the earlier Flush() fix actually meaningful.
ReadHeaderTimeout — 10s, header-only, and correctly scoped: ReadTimeout/WriteTimeout were not added, so long-lived websocket connections are unaffected. The code comment says exactly that, which should stop someone "helpfully" adding the other two later.
Local verification on the PR head:
go build ./... # ok
go vet ./cmd/... ./adapters/backend/... ./config/... # ok
go test -short ./adapters/backend/... ./config/... # ok
No blockers remaining from my side. LGTM — nice iteration on this.
|
@matthyx thanks, I will open a follow up PR adding SASL/TLS once this one is merged. |
2 participants
Overview
Adds Kafka as a pluggable message-queue backend for the synchronizer server, building on the Phase 1 messaging abstraction. Kafka is fully opt-in and selected via config - Pulsar remains the default with zero behavior change for existing deployments.
Current behavior: the server bridges WebSocket-connected clusters to a Pulsar-only backend bus.
New behavior: setting
backend.messageQueue.type: "kafka"routes the same message flow through Kafka instead. The choice is confined to thenewFromConfigfactory behind the existingMessageProducer/MessageReaderinterfaces, so the adapter, WebSocket handling, and reconciliation logic are unchanged regardless of backend.Implemented with the pure-Go, cgo-free
twmb/franz-goclient (keeps cgo-free builds, per maintainer review).Additional Information
Key design points (all from the design proposal):
.out(cluster→backend), reader →.in(backend→cluster). No self-loop filter needed.{account}/{cluster}on.out- ordering-correctness requirement so a Put and a following Delete for the same resource land on one partition, in order.group.id({groupIdPrefix}-{hostname}),auto.offset.reset=latest, auto-commit disabled → every pod sees the whole.intopic (never a competing consumer group).messaging.BuildProducerProperties); values are UTF-8, empty/absent optional metadata is omitted.BrokerMax{Write,Read}Bytesauto-raised whenmaxMessageBytesexceeds franz-go's 100 MB default.synchronizer_mq_producer_*withbackend="kafka"), and worker-pool pattern reused unchanged.Out of scope (later phases): SASL/TLS enforcement (config fields are parsed now; non-
PLAINTEXTfails loudly - enforcement is an immediate follow-up),event-ingester-serviceKafka support (Phase 3), and Helm chart plumbing (Phase 4).How to Test
Automated (needs Docker; spins up Redpanda per test):
Covers factory wiring, header +
account/clusterkey assertions, Put→Delete same-partition ordering, ~64 MB round-trip, reader→adapter dispatch, and multi-group fan-out.Manual smoke:
armo.kubescape.synchronizer.{out,in}(eachmax.message.bytes=67108864).backend.messageQueue.type: "kafka"inconfiguration/server/config.jsonwithbootstrapServers: ["localhost:9092"].LOGGER_LEVEL=debug go run ./cmd/server→ confirminitializing kafka client/kafka message queue initializedand reader startup logs.rpk topic produce armo.kubescape.synchronizer.in -H event=PutObject -H account=a1 -H cluster=c1→ server logs show the message consumed (skipped as unrelated when no cluster is connected on the pod - expected at-most-once behavior).Related issues/PRs
Checklist before requesting a review
Summary by CodeRabbit
messageQueueconfig block, with Kafka support and Pulsar fallback.