🔒 fix(sink): redact kafka SASL construction errors

internal/sink/kafka/backend.go

@@ -6,6 +6,7 @@ package kafka
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 	"time"
@@ -143,11 +144,15 @@ func dialTransport(cfg Config) (*kafka.Transport, error) {
 	if cfg.Username != "" {
 		mechanism, err := scram.Mechanism(scram.SHA256, cfg.Username, cfg.Password)
 		if err != nil {
-			return nil, fmt.Errorf("kafka SASL: %w", err)
+			return nil, redactedSASLError()
 		}
 
 		transport.SASL = mechanism
 	}
 
 	return transport, nil
 }
+
+func redactedSASLError() error {
+	return errors.New("kafka SASL: authentication configuration failed")
+}

internal/sink/kafka/sasl_redact_test.go

@@ -0,0 +1,27 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2026 Konrad Heimel
+
+package kafka
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestDialTransport_SASLPrepErrorOmitsPassword(t *testing.T) {
+	t.Parallel()
+
+	const secret = "s3cret\x07"
+	_, err := dialTransport(Config{
+		Brokers:  []string{"127.0.0.1:1"},
+		Topic:    "inventory",
+		Username: "alice",
+		Password: secret,
+	})
+	if err == nil {
+		t.Fatal("expected SASL construction error for a prohibited password rune")
+	}
+	if strings.Contains(err.Error(), "s3cret") || strings.Contains(err.Error(), secret) {
+		t.Fatalf("SASL error must not include the password: %q", err.Error())
+	}
+}