Part 1: Bigtable client and Kafka ingestion#3776
Conversation
…rt 1) Part 1 of the Bigtable historical MVCC state offload: the Bigtable data client (hand-rolled over the repo's pinned gRPC v1.57, with bounded read retries, keepalive, and billing-dimension cost metrics), the frozen row key encoding (m|shard|store|key|inverted-version mutation rows and v|bucket|inverted-version version markers), the point-read Reader, and the Kafka consumer that ingests the changelog stream into Bigtable with per-partition ordering, version dedupe, locality-chunked bulk writes, and offset commits only after durable sink writes. Connects to Google Cloud Managed Kafka via TLS + SASL/PLAIN. Part 2 wires the node-side pruned-read fallback on top of this. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
PR SummaryMedium Risk Overview A new A standalone
Reviewed by Cursor Bugbot for commit b7b8327. Bugbot is set up for automated code reviews on this repo. Configure here. |
|
The latest Buf updates on your PR. Results from workflow Buf / buf (pull_request).
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #3776 +/- ##
==========================================
- Coverage 60.11% 58.95% -1.16%
==========================================
Files 2302 2212 -90
Lines 192098 181398 -10700
==========================================
- Hits 115479 106943 -8536
+ Misses 66291 65054 -1237
+ Partials 10328 9401 -927
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
![seidroid[bot]](https://avatars.githubusercontent.com/in/2745135?s=60&v=4){kind=small}
There was a problem hiding this comment.
Part 1 of the Bigtable historical-state offload adds a well-tested, self-contained gRPC Bigtable client and a Kafka ingestion consumer with no node-behavior changes. No hard blockers, but there is a credential-exposure hardening gap (SASL/PLAIN accepted without TLS) plus two latent-correctness/observability issues (non-contiguous LastVersion, unaggregated lag gauge) worth addressing before the part-2 read path lands.
Findings: 0 blocking | 5 non-blocking | 2 posted inline
Blockers
- None at the file/PR level.
Non-blocking
REVIEW_GUIDELINES.md(from the base branch) andcursor-review.mdare both empty, so no repo-specific guidelines and no Cursor second-opinion pass were available for this synthesis.LastVersion/BigtableLastVersion(historical/bigtable.go) returns the max version marker across buckets. Because Kafka is hash-partitioned and consumed concurrently with per-partition ordering, version N's marker can be durable while lower versions on a slower partition are not yet written — so this value overstates contiguous history. Not consumed in this PR (read fallback is part 2), but if part 2 uses it to gate historical reads/pruning it needs a contiguous watermark, not a max. (Codex High — agreed, downgraded to non-blocker since it's latent in this part.)- General: the mutation-row and version-marker cell timestamps are derived from the block version (
BigtableTimestamp(version)), which is intentional for idempotent redelivery — worth a brief note/test that operators must not enable Bigtable GC-by-age on the column family, or old-height cells could be reaped. - 2 suggestion(s)/nit(s) flagged inline on specific lines.
| switch strings.ToLower(cfg.SASLMechanism) { | ||
| case "", kafkaOptionNone: | ||
| return nil | ||
| case "plain": |
There was a problem hiding this comment.
[suggestion] SASL/PLAIN sends the username and password to the broker in cleartext, but this plain case only validates that credentials are present — it never requires TLS. The aws-msk-iam case below correctly rejects !TLSEnabled; the plain case should do the same so a misconfigured producer/consumer can't leak the service-account credential over an unencrypted connection. The shipped example config and README both set TLSEnabled: true, so this is a defense-in-depth guard, but it's cheap to add:
case "plain":
if !cfg.TLSEnabled {
return fmt.Errorf("kafka tls must be enabled for sasl plain")
}
if cfg.Username == "" || cfg.Password == "" {
return fmt.Errorf("kafka username and password are required for sasl plain")
}(Codex High.)
| } | ||
| m.sinkWriteLatency.Record(ctx, writeLatency.Seconds()) | ||
| if lag >= 0 { | ||
| m.kafkaLag.Record(ctx, lag) |
There was a problem hiding this comment.
[suggestion] kafkaLag is a single unlabeled gauge, and recordBatch is called concurrently by every worker goroutine (one per shard/partition-group). Each worker overwrites the gauge with only its latest batch's lag, so it's last-writer-wins: a low-lag partition can mask another partition that is far behind, and the exported value flaps between workers. Consider a per-partition attribute on the gauge, or tracking a process-wide max/sum, so the metric reflects the worst-case backlog rather than whichever worker recorded last. (Codex Medium.)
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
| } | ||
| if bytes > 0 { | ||
| m.bytesWritten.Add(ctx, bytes, attrs) | ||
| } |
There was a problem hiding this comment.
Hot-path metric attrs reallocated
Medium Severity
recordRead and recordWrite build metric.WithAttributes(attribute.String("table", table)) on every RPC. The table name is fixed at client construction, so this allocates on the ingestion and read hot paths. Existing SeiDB metrics (for example CacheMetrics) precompute the attribute option once at init and reuse it.
Triggered by learned rule: OTel metrics: guard attribute cardinality and use native types
Reviewed by Cursor Bugbot for commit 0c5d3dd. Configure here.
| grpc.WithKeepaliveParams(keepalive.ClientParameters{ | ||
| Time: 30 * time.Second, | ||
| Timeout: 10 * time.Second, | ||
| }), |
There was a problem hiding this comment.
Idle keepalive pings disabled
Low Severity
The client sets gRPC keepalive Time/Timeout specifically to detect silently dropped Bigtable connections, but omits PermitWithoutStream: true. Without that flag, keepalive pings are only sent while a stream is active, so idle connections between consumer batches can still die quietly until the next RPC fails.
Reviewed by Cursor Bugbot for commit 0c5d3dd. Configure here.
Kbhat1
changed the title
Kbhat1
changed the title
Kbhat1
changed the title
Kbhat1
changed the title
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
|
|
||
| switch strings.ToLower(c.SASLMechanism) { | ||
| return ValidateSASL(*c) | ||
| } | ||
|
|
||
| // ValidateSASL checks the SASL mechanism and the credentials it requires. | ||
| // Shared by the producer and the offload consumer configs. | ||
| func ValidateSASL(cfg KafkaConfig) error { | ||
| switch strings.ToLower(cfg.SASLMechanism) { | ||
| case "", kafkaOptionNone: | ||
| return nil | ||
| case "plain": | ||
| if cfg.Username == "" || cfg.Password == "" { | ||
| return fmt.Errorf("kafka username and password are required for sasl plain") | ||
| } | ||
| case "aws-msk-iam": | ||
| if !c.TLSEnabled { | ||
| if !cfg.TLSEnabled { | ||
| return fmt.Errorf("kafka tls must be enabled for aws-msk-iam") | ||
| } | ||
| if c.Region == "" { | ||
| if cfg.Region == "" { |
There was a problem hiding this comment.
🔴 ValidateSASL's "plain" case in sei-db/state_db/ss/offload/kafka.go only checks that Username/Password are non-empty and never requires cfg.TLSEnabled, unlike the aws-msk-iam case a few lines below which does. This lets an operator configure SASLMechanism="plain" with TLSEnabled=false, causing NewKafkaStream/NewKafkaReader to dial a plaintext TCP connection while sending SASL/PLAIN credentials (e.g. a GCP service-account key, per the consumer README) in the clear, since PLAIN is base64-framed but not encrypted. Fix by requiring TLSEnabled in the "plain" branch, mirroring the aws-msk-iam check.
Extended reasoning...
This PR adds a new "plain" case to the shared ValidateSASL function in sei-db/state_db/ss/offload/kafka.go to support Google Cloud Managed Kafka authentication. The new case validates only that Username and Password are non-empty:
case "plain":
if cfg.Username == "" || cfg.Password == "" {
return fmt.Errorf("kafka username and password are required for sasl plain")
}The very next case in the same switch statement, "aws-msk-iam", explicitly requires TLS:
case "aws-msk-iam":
if !cfg.TLSEnabled {
return fmt.Errorf("kafka tls must be enabled for aws-msk-iam")
}SASL/PLAIN transmits the username and password base64-encoded in the SASL exchange, which is a framing, not an encryption. Anyone who can observe the TCP connection can trivially decode the base64 payload and recover the credentials. TLS is what actually protects the credentials on the wire for PLAIN auth — this is precisely why the codebase already enforces TLSEnabled for aws-msk-iam right below it. The PR does not carry that same discipline over to the newly-added plain mechanism.
Code path: Both NewKafkaStream (producer, existing) and NewKafkaReader (new consumer, sei-db/state_db/ss/offload/consumer/kafka.go) gate TLS strictly on cfg.TLSEnabled:
if cfg.TLSEnabled {
dialer.TLS = &tls.Config{MinVersion: tls.VersionTLS12}
}ValidateSASL runs before this and does not block a config where SASLMechanism == "plain" and TLSEnabled == false. So such a config passes validation, and the dialer/mechanism proceed to dial plaintext TCP while still attaching plain.Mechanism{Username, Password} to the connection — sending the credentials over an unencrypted channel.
Step-by-step proof:
- Operator sets
Kafka.TLSEnabled = false(default value if omitted) andKafka.SASLMechanism = "plain"withUsername/Passwordpopulated (e.g. following the README's Google Cloud Managed Kafka example, wherePasswordis a base64-encoded GCP service-account key JSON). KafkaReaderConfig.Validate()callsoffload.ValidateSASL(c.saslConfig()), which only checksUsername != "" && Password != ""for"plain"and returnsnil— no error.NewKafkaReader(orNewKafkaStream) proceeds: sincecfg.TLSEnabledisfalse,dialer.TLSis leftnil, so the connection to the Kafka brokers is a plaintext TCP socket.NewSASLMechanismstill returnsplain.Mechanism{Username, Password}and attaches it to the dialer/transport, so the SASL PLAIN handshake — carrying the base64-framed (not encrypted) credentials — is sent over that plaintext socket.- Any on-path network observer (e.g. anyone with visibility into the VPC/hosting network) can capture the TCP stream and base64-decode the SASL PLAIN frame to recover the GCP service-account credentials.
This is confirmed directly by the PR's own test, TestKafkaReaderConfigValidateSASL in sei-db/state_db/ss/offload/consumer/kafka_test.go: it exercises plain with Username/Password set and never sets TLSEnabled, and asserts cfg.Validate() returns no error — i.e., the test locks in the gap rather than catching it.
Fix: add a TLS check in the "plain" branch, e.g.:
case "plain":
if !cfg.TLSEnabled {
return fmt.Errorf("kafka tls must be enabled for sasl plain")
}
if cfg.Username == "" || cfg.Password == "" {
return fmt.Errorf("kafka username and password are required for sasl plain")
}This mirrors the existing aws-msk-iam guard exactly and is a one-line addition. The shipped example config (config/example-bigtable.json) already sets TLSEnabled: true, so the fix does not affect the documented happy path — it only closes a misconfiguration gap that would otherwise leak credentials.
| m.sinkWriteLatency.Record(ctx, writeLatency.Seconds()) | ||
| if lag >= 0 { | ||
| m.kafkaLag.Record(ctx, lag) | ||
| } |
There was a problem hiding this comment.
Lag gauge races across workers
Medium Severity
consumer_kafka_lag is a single process-wide gauge updated from every worker after each batch. With concurrent workers, last writer wins, so a low-lag partition can overwrite a high-lag one and the scraped value no longer reflects true consumer lag.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit bcde772. Configure here.
| "consumer_kafka_lag", | ||
| metric.WithDescription("Messages between the last processed offset and the partition high watermark (max across the batch)"), | ||
| metric.WithUnit("{message}"), | ||
| ) |
There was a problem hiding this comment.
Consumer metrics lack scope prefix
Low Severity
Prometheus series use short names like consumer_records_processed_total and consumer_kafka_lag. The meter seidb_offload_consumer does not become a series prefix, so these names are underscoped relative to repo convention (seiwal_*, pebblecache_*, bigtable_*) and the OTel naming rule.
Triggered by learned rule: OTel Prometheus export: namespace prefix and unit auto-suffix naming
Reviewed by Cursor Bugbot for commit bcde772. Configure here.
There was a problem hiding this comment.
A well-structured, well-tested Part 1 that adds a self-contained Bigtable client and a Kafka→Bigtable ingestion consumer with no node-behavior change. No blocking issues; a few robustness/design notes are worth addressing before Part 2 wires in the read fallback.
Findings: 0 blocking | 7 non-blocking | 3 posted inline
Blockers
- None at the file/PR level.
Non-blocking
- Cursor's review file (cursor-review.md) was empty — that pass produced no output, so its perspective is not reflected here.
- Merged from Codex:
LastVersionreturns the max version marker, but because versions are ingested concurrently across partition-sharded workers, a higher version can be marked durable before lower ones finish.LastVersionis therefore a high-water-mark, not a 'complete up to V' guarantee. Harmless in this PR (unused by node code) but the contract should be documented/tightened before Part 2's read fallback relies on it to decide the local-SS retention boundary — otherwise pruned-but-not-yet-ingested versions could read as absent. - Merged from Codex: decoded changelog entries are never structurally validated. A malformed/corrupt message with a negative
VersionreachesbigtableInvertedVersion→uint64FromNonNegativeInt64, which panics; an unrecovered panic in a worker goroutine crashes the whole consumer. Since offsets commit only after a successful write, a single poison message becomes a restart/crash loop. The input stream is the node's own trusted changelog so this is unlikely in practice, but converting these encode-time panics into per-message errors (poison-message handling / dead-letter or skip-with-log) would make ingestion robust.compactMutationsalso assumes non-nilChangesets/Pairselements. - The example config documents storing a base64 service-account key as the Kafka
Passwordin a plaintext JSON file. Consider recommending a secrets manager / env indirection in the README rather than a key committed to an operator's config file. - 3 suggestion(s)/nit(s) flagged inline on specific lines.
| return bigtableValueFromRow(row, r.family) | ||
| } | ||
|
|
||
| func BigtableLastVersion(ctx context.Context, readRows BigtableReadRowsFunc) (int64, error) { |
There was a problem hiding this comment.
[suggestion] LastVersion returns max(bucketVersion) across all buckets. Because consecutive versions are ingested concurrently by partition-sharded workers, a higher version's marker can be written before a lower version finishes, so this value does not guarantee every version below it is present. That's fine here (nothing consumes it yet), but Part 2's read fallback will presumably use it to decide which versions are safely offloaded — please document that this is a high-water-mark, not a contiguity guarantee, or track a distinct 'contiguously-complete' watermark before relying on it for pruning decisions.
| records := make([]Record, 0, len(msgs)) | ||
| var firstVersion, lastVersion int64 | ||
| for i, msg := range msgs { | ||
| entry, err := DecodeEntry(msg.Value) |
There was a problem hiding this comment.
[suggestion] Decoded entries are used without structural validation. A corrupt message with a negative Version flows into BigtableMutationRowKey → bigtableInvertedVersion → uint64FromNonNegativeInt64, which panics; an unrecovered panic in the worker goroutine crashes the process. Because offsets are committed only after a successful write, the same poison message is re-fetched on restart → crash loop. Consider validating the entry here (non-negative version, non-nil changesets/pairs) and returning it as a bounded/skippable error instead of letting encoding panic.
There was a problem hiding this comment.
+1, If a Kafka message contains invalid protobuf, DecodeEntry returns an error, processBatch returns the error, the worker kills the errgroup, Run returns, and main calls log.Fatalf. On restart, the consumer re-fetches the same uncommitted message and crashes again — infinite loop. There's no dead-letter mechanism, no skip-after-N-failures, no offset advance for undecodable messages.
| // followed by a tombstone leaves both cells on the row. | ||
| func (s *bigtableSink) mutationRow(version int64, storeName string, pair *proto.KVPair) historical.BigtableRowMutation { | ||
| ts := historical.BigtableTimestamp(version) | ||
| deleted := pair.Delete || pair.Value == nil |
There was a problem hiding this comment.
[nit] deleted := pair.Delete || pair.Value == nil conflates a live write of a nil value with a tombstone, so a genuine nil-value write is stored (and later read) as absent. bigtableValueFromRow mirrors this (value == nil → ErrNotFound), so it's self-consistent, but worth a comment confirming SS never emits a live nil-value pair — otherwise such writes silently disappear.
| } | ||
| } | ||
|
|
||
| func sleepWithContext(ctx context.Context, d time.Duration) error { |
There was a problem hiding this comment.
Identical implementations in both consumer/consumer.go and historical/bigtable.go. Extract to a shared util package.
| "Username": "kafka-client@my-gcp-project.iam.gserviceaccount.com", | ||
| "Password": "<base64-encoded service account key JSON>" |
There was a problem hiding this comment.
Kafka SASL passwords live in a plain JSON file on disk. No secret manager integration, no envelope encryption. For a production system handling chain state, this is a security concern. At minimum, support reading credentials from environment variables or a secrets manager.
|
|
||
| const ( | ||
| DefaultBigtableFamily = "state" | ||
| DefaultBigtableShards = 256 |
There was a problem hiding this comment.
The shard value is baked into every mutation row key: FNV32a(storeName + \0 + key) % Shards. If an operator changes Shards from 256 to 512, every Get and Has query computes a different shard for the same (storeName, key), so it looks up the wrong row prefix and returns ErrNotFound. All previously ingested data becomes unreachable with no error. This needs to either be persisted in a metadata row and validated on startup, or enforced as immutable in the config.
| if len(errs) != len(rows) { | ||
| return fmt.Errorf("bigtable returned %d mutation results for %d rows", len(errs), len(rows)) | ||
| } | ||
| for i, rowErr := range errs { |
There was a problem hiding this comment.
If 50 out of 1024 rows fail, you see one error and no indication of how widespread the failure was. This makes it much harder to distinguish "one hot tablet is overloaded" from "everything is broken."
Consider Include the failure count:
return fmt.Errorf("%d of %d rows failed, first: row %q: %w", failCount, len(rows), rows[firstIdx].RowKey, errs[firstIdx])
| func (c *Consumer) writeBatchWithRetry(ctx context.Context, records []Record) error { | ||
| backoff := sinkBaseBackoff | ||
| var lastErr error | ||
| for attempt := 1; attempt <= sinkMaxAttempts; attempt++ { |
There was a problem hiding this comment.
writeBatchWithRetry retries the entire batch even after partial success, If 1023 out of 1024 mutation rows succeed and 1 fails, the retry re-sends all 1024. Since SetCell is idempotent at the same timestamp this is correct, but it amplifies write cost and latency proportional to batch size. A targeted retry of only failed rows would be more efficient — ApplyBulk already returns per-row errors.
| records := make([]Record, 0, len(msgs)) | ||
| var firstVersion, lastVersion int64 | ||
| for i, msg := range msgs { | ||
| entry, err := DecodeEntry(msg.Value) |
There was a problem hiding this comment.
+1, If a Kafka message contains invalid protobuf, DecodeEntry returns an error, processBatch returns the error, the worker kills the errgroup, Run returns, and main calls log.Fatalf. On restart, the consumer re-fetches the same uncommitted message and crashes again — infinite loop. There's no dead-letter mechanism, no skip-after-N-failures, no offset advance for undecodable messages.
| if err := c.Kafka.Validate(); err != nil { | ||
| return fmt.Errorf("kafka: %w", err) | ||
| } | ||
| bigtable := c.Bigtable |
There was a problem hiding this comment.
BigtableConfig is a struct (not a pointer), so bigtable := c.Bigtable creates a copy.
ApplyDefaults() fills in Family = "state" and Shards = 256 on the copy. Then Validate() passes because the copy has those defaults. But the original c.Bigtable is untouched — c.Bigtable.Family is still "".
It happens to work in practice because NewBigtableSink calls cfg.ApplyDefaults() again before using it. But if any code between LoadConfig and NewBigtableSink reads cfg.Bigtable.Family (e.g., for logging or metrics tagging), it sees an empty string despite validation having passed.
The fix is simple — apply defaults on the original, not a copy.
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 4 potential issues.
There are 8 total unresolved issues (including 4 from previous reviews).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Want higher recall? High effort reviews run extra passes and find more bugs. A team admin can switch effort levels in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit b7b8327. Configure here.
| for i, msg := range msgs { | ||
| entry, err := DecodeEntry(msg.Value) | ||
| if err != nil { | ||
| return fmt.Errorf("decode message at offset %d: %w", msg.Offset, err) |
There was a problem hiding this comment.
Poison messages crash the consumer
High Severity
DecodeEntry failures return from processBatch, kill the errgroup, and main calls log.Fatalf without committing offsets. On restart the same uncommitted message is fetched again, causing a crash loop. A negative Version also panics in uint64FromNonNegativeInt64 with the same outcome. There is no dead-letter, skip-after-N, or offset advance for poison payloads.
Additional Locations (2)
Reviewed by Cursor Bugbot for commit b7b8327. Configure here.
| } | ||
| if c.Shards == 0 { | ||
| c.Shards = DefaultBigtableShards | ||
| } |
There was a problem hiding this comment.
Shard count silently orphans data
High Severity
Shards is part of every mutation row key via FNV32a(storeName + \0 + key) % Shards, but it is only a free config field with a default. Changing Shards after ingestion makes Get/Has compute a different shard prefix and return ErrNotFound, so previously written history becomes unreachable with no error. Nothing persists or validates the shard count against the table.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit b7b8327. Configure here.
| bigtable.ApplyDefaults() | ||
| if err := bigtable.Validate(); err != nil { | ||
| return fmt.Errorf("bigtable: %w", err) | ||
| } |
There was a problem hiding this comment.
Bigtable defaults applied to copy
Medium Severity
Validate copies c.Bigtable before ApplyDefaults, so validation succeeds with default Family/Shards while the original config still has Family == "" and Shards == 0. LoadConfig therefore returns an un-defaulted Bigtable even after a successful validate. This only works today because NewBigtableSink applies defaults again.
Reviewed by Cursor Bugbot for commit b7b8327. Configure here.
| // "aws-msk-iam". | ||
| SASLMechanism string | ||
| Username string | ||
| Password string |
There was a problem hiding this comment.
Plaintext Kafka credentials on disk
Medium Severity
SASL Username/Password are loaded only from a plain JSON config file. For Managed Kafka this password is a base64 service-account key. There is no env-var or secret-manager path, so production deployments are pushed toward storing long-lived credentials on disk in cleartext.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit b7b8327. Configure here.
There was a problem hiding this comment.
Part 1 of the Bigtable historical-offload work: a self-contained Bigtable gRPC client, MVCC row-key encoding, and a Kafka→Bigtable ingestion consumer with per-partition ordering, redelivery dedupe, and offset-after-write semantics. Code is well-tested and cleanly structured, but the shared SASL validation now accepts SASL/PLAIN with no TLS requirement, allowing Kafka credentials to be sent in cleartext.
Findings: 1 blocking | 3 non-blocking | 1 posted inline
Blockers
- None at the file/PR level.
- 1 blocking issue(s) flagged inline on specific lines.
Non-blocking
BigtableLastVersion(sei-db/state_db/ss/offload/historical/bigtable.go:412) returns the maximum version marker across buckets. Because the consumer shards by partition and writes concurrently, and version markers are keyed only by version (no partition), a higher version can be marked durable before a lower one's mutation rows land.LastVersioncan therefore advertise a max that is ahead of a contiguous, fully-written prefix. No node-behavior impact in this PR (the read fallback lands in part 2), but before part 2 uses this as a retention/availability boundary, consider tracking a contiguous low-watermark rather than the max, or document that LastVersion is not a completeness guarantee. (Raised by Codex.)- The Cursor second-opinion review file (
cursor-review.md) was empty — that pass produced no output. - Consumer/consumer.go retries
sink.WriteBatchup to 5 attempts with backoff and, on final failure, returns an error that abortsRun; this is intentional (offsets are only committed after a durable write) but means a persistently failing sink halts the whole consumer. Confirm that's the desired operational behavior (crash-and-restart) vs. per-shard isolation.
| switch strings.ToLower(cfg.SASLMechanism) { | ||
| case "", kafkaOptionNone: | ||
| return nil | ||
| case "plain": |
There was a problem hiding this comment.
[blocker] SASL/PLAIN is accepted here with no TLS requirement. SASL/PLAIN transmits the username and password effectively in cleartext, so a plain config with TLSEnabled: false leaks Kafka credentials on the wire. This is now reachable on the producer path too (NewKafkaStream → KafkaConfig.Validate → ValidateSASL), not just the new consumer. The aws-msk-iam case already requires TLS — apply the same guard here:
case "plain":
if !cfg.TLSEnabled {
return fmt.Errorf("kafka tls must be enabled for sasl plain")
}
if cfg.Username == "" || cfg.Password == "" {
return fmt.Errorf("kafka username and password are required for sasl plain")
}(Flagged by Codex as P1.)
2 participants
Description
Testing