From 746f75ab0f786cb92a791444a0add964731f9883 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sat, 3 Oct 2026 22:32:08 +0800 Subject: [PATCH] fix(kafka): retain failed message persistence --- deploy/argocd/im-server-test/README.md | 20 +++++++ .../statefulset-performance-patch.yaml | 9 +++ docker/tinode/config.template | 2 +- server/kafka_audit_test.go | 4 ++ server/kafka_message_persist.go | 24 ++++---- server/kafka_runtime.go | 2 +- server/kafka_runtime_batch_test.go | 57 ++++++++++++------- server/meetingim.conf | 2 +- server/tinode.conf | 2 +- 9 files changed, 85 insertions(+), 37 deletions(-) diff --git a/deploy/argocd/im-server-test/README.md b/deploy/argocd/im-server-test/README.md index dfe91ca..38b2f12 100644 --- a/deploy/argocd/im-server-test/README.md +++ b/deploy/argocd/im-server-test/README.md @@ -25,6 +25,7 @@ After ArgoCD sync, verify the effective configuration: kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].resources}' kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].livenessProbe.httpGet.path}' kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].env}' +kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.dnsConfig.options}' kubectl auth can-i list pods.metrics.k8s.io \ --as=system:serviceaccount:argocd:backend-debug \ -n im-server-system @@ -36,5 +37,24 @@ liveness path `/livez`. The MySQL pool environment must be `MYSQL_REGISTRY_MAX_OPEN_CONNS=4`, and `MYSQL_REGISTRY_MAX_IDLE_CONNS=2`. The WebSocket heartbeat environment must be `WS_PING_PERIOD_SECONDS=15` and `WS_PONG_WAIT_SECONDS=90`. +The Pod DNS options must be `ndots:1`, `timeout:2`, and `attempts:2`. Keeping +`ClusterFirst` preserves short Kubernetes service-name resolution, while the +lower `ndots` value prevents fully qualified internal and external dependency +names from being expanded through every cluster search suffix first. The application repository's deploy workflow waits for reconciliation and fails if ArgoCD restores different values. + +After the rollout, verify the generated resolver configuration on every Pod: + +```bash +for pod in im-server-0 im-server-1 im-server-2; do + echo "--- ${pod}" + kubectl -n im-server-system exec "${pod}" -- cat /etc/resolv.conf +done +``` + +The output must retain the cluster DNS nameserver and search domains, with +`options ndots:1 timeout:2 attempts:2`. This mitigation reduces amplified DNS +traffic from the application Pods. If lookup timeouts remain under load, +CoreDNS capacity or NodeLocal DNSCache must be changed in the cluster +infrastructure repository by an administrator with `kube-system` access. diff --git a/deploy/argocd/im-server-test/statefulset-performance-patch.yaml b/deploy/argocd/im-server-test/statefulset-performance-patch.yaml index a68060c..b10aeb4 100644 --- a/deploy/argocd/im-server-test/statefulset-performance-patch.yaml +++ b/deploy/argocd/im-server-test/statefulset-performance-patch.yaml @@ -6,6 +6,15 @@ metadata: spec: template: spec: + dnsPolicy: ClusterFirst + dnsConfig: + options: + - name: ndots + value: "1" + - name: timeout + value: "2" + - name: attempts + value: "2" containers: - name: im-server env: diff --git a/docker/tinode/config.template b/docker/tinode/config.template index 5443833..53a1025 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -135,7 +135,7 @@ "write_timeout_ms": 3000, "read_timeout_ms": 3000, "batch_timeout_ms": 10, - "required_acks": 1, + "required_acks": -1, "allow_auto_topic_creation": false, "tls_enabled": false, "sasl_enabled": false, diff --git a/server/kafka_audit_test.go b/server/kafka_audit_test.go index 28155b4..09687bd 100644 --- a/server/kafka_audit_test.go +++ b/server/kafka_audit_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/golang/mock/gomock" + "github.com/segmentio/kafka-go" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/mock_store" "github.com/tinode/chat/server/store/types" @@ -39,6 +40,9 @@ func TestNormalizeKafkaConfigUsesLowLatencyBatchTimeout(t *testing.T) { if cfg.MaxRetryCount != 3 { t.Fatalf("unexpected kafka max retry count: %d", cfg.MaxRetryCount) } + if cfg.RequiredAcks != int(kafka.RequireAll) { + t.Fatalf("unexpected kafka required acks: %d", cfg.RequiredAcks) + } } func TestEnqueueKafkaAuditEventPublishesCrossSystemEvent(t *testing.T) { diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go index 71025eb..b6ae287 100644 --- a/server/kafka_message_persist.go +++ b/server/kafka_message_persist.go @@ -281,27 +281,27 @@ func persistKafkaMessageIndividually(ctx context.Context, item *kafkaMessagePers return store.HTTPInfrastructure.MarkKafkaConsumeFailed(item.tenantID, item.envelope.EventID, kafkaBus.config.ConsumerGroup, kafkaMessagePersistFailureReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond)) } - markDead := func(failed *types.KafkaConsumeRecord, err error) error { - immetrics.KafkaMessagePersistTotal.WithLabelValues(kind, "dead").Inc() - immetrics.MessagePersistDuration.WithLabelValues(kind, "dead").Observe(time.Since(started).Seconds()) - reason := kafkaMessagePersistFailureReason(err) - immetrics.KafkaDeadLettersTotal.WithLabelValues(kind, reason).Inc() - return markKafkaDeadLetter(ctx, item.tenantID, item.envelope, failed, reason, err.Error()) + err := runKafkaMessageSingleRetries(ctx, kafkaBus.config.MessageRetries, operation, markFailed, waitKafkaRetry) + if err != nil { + // Message persistence failures are recoverable infrastructure failures. Returning + // the error keeps the Kafka offset uncommitted so the outer consumer retries the + // message until it is durable. Only malformed payloads are dead-lettered while + // preparing the item. + immetrics.KafkaMessagePersistTotal.WithLabelValues(kind, "retry_exhausted").Inc() + immetrics.MessagePersistDuration.WithLabelValues(kind, "retry_exhausted").Observe(time.Since(started).Seconds()) } - return runKafkaMessageSingleRetries(ctx, kafkaBus.config.MessageRetries, operation, markFailed, markDead, waitKafkaRetry) + return err } func runKafkaMessageSingleRetries(ctx context.Context, retryCount int, operation func() error, - markFailed func(error) (*types.KafkaConsumeRecord, error), markDead func(*types.KafkaConsumeRecord, error) error, - wait kafkaRetryWaiter) error { + markFailed func(error) (*types.KafkaConsumeRecord, error), wait kafkaRetryWaiter) error { var lastErr error - var failed *types.KafkaConsumeRecord for attempt := 1; attempt <= retryCount+1; attempt++ { if lastErr = operation(); lastErr == nil { return nil } var err error - if failed, err = markFailed(lastErr); err != nil { + if _, err = markFailed(lastErr); err != nil { return err } if attempt <= retryCount { @@ -310,7 +310,7 @@ func runKafkaMessageSingleRetries(ctx context.Context, retryCount int, operation } } } - return markDead(failed, lastErr) + return fmt.Errorf("message persistence retries exhausted: %w", lastErr) } func finishKafkaMessagePersist(ctx context.Context, item *kafkaMessagePersistItem) error { diff --git a/server/kafka_runtime.go b/server/kafka_runtime.go index 1070725..6ed60ec 100644 --- a/server/kafka_runtime.go +++ b/server/kafka_runtime.go @@ -182,7 +182,7 @@ func normalizeKafkaConfig(cfg *kafkaConfig) { cfg.BatchTimeoutMS = 10 } if cfg.RequiredAcks == 0 { - cfg.RequiredAcks = int(kafka.RequireOne) + cfg.RequiredAcks = int(kafka.RequireAll) } if cfg.MaxRetryCount <= 0 { cfg.MaxRetryCount = 3 diff --git a/server/kafka_runtime_batch_test.go b/server/kafka_runtime_batch_test.go index 1ac7533..e79c847 100644 --- a/server/kafka_runtime_batch_test.go +++ b/server/kafka_runtime_batch_test.go @@ -3,6 +3,7 @@ package main import ( "context" "errors" + "strings" "testing" "time" @@ -47,6 +48,33 @@ func TestConsumeKafkaMessageBatchRetriesHandlerAndCommit(t *testing.T) { } } +func TestConsumeKafkaMessageBatchDoesNotCommitWhileHandlerFails(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + handlerCalls := 0 + commitCalls := 0 + + err := consumeKafkaMessageBatch(ctx, []kafka.Message{{Offset: 1}}, + func(context.Context, []kafka.Message) error { + handlerCalls++ + return errors.New("database unavailable") + }, + func(context.Context, ...kafka.Message) error { + commitCalls++ + return nil + }, + func(ctx context.Context, _ time.Duration) error { + cancel() + return ctx.Err() + }) + + if !errors.Is(err, context.Canceled) { + t.Fatalf("consumeKafkaMessageBatch() error = %v, want context canceled", err) + } + if handlerCalls != 1 || commitCalls != 0 { + t.Fatalf("calls handler=%d commit=%d, want 1/0", handlerCalls, commitCalls) + } +} + func TestConsumeKafkaMalformedMessageRetriesCommit(t *testing.T) { commitCalls := 0 waitCalls := 0 @@ -76,10 +104,9 @@ func TestConsumeKafkaMalformedMessageRetriesCommit(t *testing.T) { } } -func TestRunKafkaMessageSingleRetriesThenDeadLetters(t *testing.T) { +func TestRunKafkaMessageSingleRetriesReturnsErrorWithoutTerminalCommit(t *testing.T) { operationCalls := 0 markFailedCalls := 0 - markDeadCalls := 0 waitCalls := 0 err := runKafkaMessageSingleRetries(context.Background(), 3, @@ -91,31 +118,23 @@ func TestRunKafkaMessageSingleRetriesThenDeadLetters(t *testing.T) { markFailedCalls++ return &types.KafkaConsumeRecord{RetryCount: markFailedCalls}, nil }, - func(record *types.KafkaConsumeRecord, err error) error { - markDeadCalls++ - if record == nil || record.RetryCount != 4 || err == nil { - t.Fatalf("unexpected dead-letter input: record=%+v err=%v", record, err) - } - return nil - }, func(context.Context, time.Duration) error { waitCalls++ return nil }) - if err != nil { - t.Fatalf("runKafkaMessageSingleRetries() error = %v", err) + if err == nil || !strings.Contains(err.Error(), "message persistence retries exhausted") { + t.Fatalf("runKafkaMessageSingleRetries() error = %v, want retry exhaustion", err) } - if operationCalls != 4 || markFailedCalls != 4 || waitCalls != 3 || markDeadCalls != 1 { - t.Fatalf("calls operation=%d failed=%d wait=%d dead=%d, want 4/4/3/1", - operationCalls, markFailedCalls, waitCalls, markDeadCalls) + if operationCalls != 4 || markFailedCalls != 4 || waitCalls != 3 { + t.Fatalf("calls operation=%d failed=%d wait=%d, want 4/4/3", + operationCalls, markFailedCalls, waitCalls) } } func TestRunKafkaMessageSingleRetriesStopsOnSuccess(t *testing.T) { operationCalls := 0 markFailedCalls := 0 - markDeadCalls := 0 err := runKafkaMessageSingleRetries(context.Background(), 3, func() error { operationCalls++ @@ -128,16 +147,12 @@ func TestRunKafkaMessageSingleRetriesStopsOnSuccess(t *testing.T) { markFailedCalls++ return &types.KafkaConsumeRecord{RetryCount: markFailedCalls}, nil }, - func(*types.KafkaConsumeRecord, error) error { - markDeadCalls++ - return nil - }, func(context.Context, time.Duration) error { return nil }) if err != nil { t.Fatalf("runKafkaMessageSingleRetries() error = %v", err) } - if operationCalls != 3 || markFailedCalls != 2 || markDeadCalls != 0 { - t.Fatalf("calls operation=%d failed=%d dead=%d, want 3/2/0", operationCalls, markFailedCalls, markDeadCalls) + if operationCalls != 3 || markFailedCalls != 2 { + t.Fatalf("calls operation=%d failed=%d, want 3/2", operationCalls, markFailedCalls) } } diff --git a/server/meetingim.conf b/server/meetingim.conf index b603150..3357731 100644 --- a/server/meetingim.conf +++ b/server/meetingim.conf @@ -122,7 +122,7 @@ "write_timeout_ms": 3000, "read_timeout_ms": 3000, "batch_timeout_ms": 10, - "required_acks": 1, + "required_acks": -1, "allow_auto_topic_creation": false, "tls_enabled": false, "sasl_enabled": true, diff --git a/server/tinode.conf b/server/tinode.conf index 350af04..eb2bba2 100644 --- a/server/tinode.conf +++ b/server/tinode.conf @@ -121,7 +121,7 @@ "write_timeout_ms": 3000, "read_timeout_ms": 3000, "batch_timeout_ms": 10, - "required_acks": 1, + "required_acks": -1, "allow_auto_topic_creation": false, "tls_enabled": false, "sasl_enabled": true,