pro/im_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(kafka): retain failed message persistence

Browse Source
This commit is contained in:
richjohnrambo
2026-10-03 22:32:08 +08:00
parent db92c41241
commit 746f75ab0f
9 changed files with 85 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/kafka_audit_test.go
+4
View File
@@ -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) {
server/kafka_message_persist.go
+12 -12
View File
@@ -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 {
server/kafka_runtime.go
+1 -1
View File
@@ -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
server/kafka_runtime_batch_test.go
+36 -21
View File
@@ -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)
}
}
server/meetingim.conf
+1 -1
View File
@@ -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,
server/tinode.conf
+1 -1
View File
@@ -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,