diff --git a/docker/tinode/Dockerfile b/docker/tinode/Dockerfile index 78e8dd6..91dbe14 100644 --- a/docker/tinode/Dockerfile +++ b/docker/tinode/Dockerfile @@ -87,6 +87,9 @@ ENV KAFKA_SASL_USERNAME= ENV KAFKA_SASL_PASSWORD= ENV KAFKA_MAX_RETRY_COUNT=3 ENV KAFKA_RETRY_BACKOFF_MS='[10000,30000,60000,300000,900000]' +ENV KAFKA_MESSAGE_BATCH_SIZE=100 +ENV KAFKA_MESSAGE_BATCH_WAIT_MS=10 +ENV KAFKA_MESSAGE_SINGLE_RETRY_COUNT=3 ENV KAFKA_DEAD_LETTER_RETENTION_SECONDS=7776000 ENV KAFKA_TOPIC_ADMIN_COMMAND=im.admin.command ENV KAFKA_TOPIC_ADMIN_COMMAND_RESULT=im.admin.command.result diff --git a/docker/tinode/Dockerfile.build b/docker/tinode/Dockerfile.build index 0ac9849..efa50f2 100644 --- a/docker/tinode/Dockerfile.build +++ b/docker/tinode/Dockerfile.build @@ -67,6 +67,9 @@ ENV KAFKA_SASL_USERNAME= ENV KAFKA_SASL_PASSWORD= ENV KAFKA_MAX_RETRY_COUNT=3 ENV KAFKA_RETRY_BACKOFF_MS='[10000,30000,60000,300000,900000]' +ENV KAFKA_MESSAGE_BATCH_SIZE=100 +ENV KAFKA_MESSAGE_BATCH_WAIT_MS=10 +ENV KAFKA_MESSAGE_SINGLE_RETRY_COUNT=3 ENV KAFKA_DEAD_LETTER_RETENTION_SECONDS=7776000 ENV KAFKA_TOPIC_ADMIN_COMMAND=im.admin.command ENV KAFKA_TOPIC_ADMIN_COMMAND_RESULT=im.admin.command.result diff --git a/docker/tinode/config.template b/docker/tinode/config.template index 49cb2f0..9e0e6c9 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -140,6 +140,9 @@ "sasl_password": "", "max_retry_count": 3, "retry_backoff_ms": [10000, 30000, 60000, 300000, 900000], + "message_batch_size": $KAFKA_MESSAGE_BATCH_SIZE, + "message_batch_wait_ms": $KAFKA_MESSAGE_BATCH_WAIT_MS, + "message_single_retry_count": $KAFKA_MESSAGE_SINGLE_RETRY_COUNT, "dead_letter_retention_seconds": 7776000, "topics": { "admin_command": "im.admin.command", diff --git a/docs/IM-Kafka接入方案.md b/docs/IM-Kafka接入方案.md index 6c7402b..8944100 100644 --- a/docs/IM-Kafka接入方案.md +++ b/docs/IM-Kafka接入方案.md @@ -724,11 +724,19 @@ IM Server 已接入 `im.message.persist`，用于把在线发送链路和最终 1. WS/HTTP 发送入口先完成权限校验，并由 topic 内存态分配 `seq_id`。 2. Kafka 开启时，私聊发送入口直接生产 `im.message.persist`，群聊发送入口直接生产 `im.group.message.persist`，不查库、不落库、不写 DB outbox；客户端在 Kafka 写入成功后拿到已分配的 `seq_id`。 3. Kafka 关闭时，继续走原同步落库路径，兼容本地和未接 Kafka 环境。 -4. 消费者按 `message_kind` 区分私聊与群聊： - - `p2p` 写入 `im_messages`。 - - `group` 写入 `im_group_messages`。 +4. 私聊和群聊使用独立 Topic、独立消费组和独立物理表：私聊批量写入租户分表 `im_%03d_messages`，群聊按租户和月份批量写入 `im_%03d_group_messages_YYYYMM`。 5. 消费侧复用 `im_kafka_consume_log` 做 `event_id + consumer_group` 幂等。 -6. 失败按统一退避重试 3 次，超过上限先投递 `im.dead.letter`，再由 dead letter consumer 写入 `im_kafka_dead_letters`。 +6. 消费者最多聚合 `message_batch_size` 条消息，在 `message_batch_wait_ms` 时间窗内形成批次并执行一条多值 `INSERT`。 +7. 批量写入失败后逐条降级；每条执行首次写入和 `message_single_retry_count` 次重试，默认共尝试 4 次。 +8. 单条仍失败时，完成 `im.dead.letter` 投递和 `im_kafka_dead_letters` 终态写入后才提交该批 offset；提交失败会持续重试，避免跳过失败消息。 + +消息批量落库参数： + +| 配置项 | 默认值 | 说明 | +| --- | ---: | --- | +| `message_batch_size` | `100` | 单批最大消息数，上限 500。 | +| `message_batch_wait_ms` | `10` | 从取得第一条消息开始的批次聚合等待时间。 | +| `message_single_retry_count` | `3` | 批量失败后单条额外重试次数，不含首次执行。 | ### 11.10 IM Server 已接入的缓存失效消费 diff --git a/docs/IM-Kafka接入联调测试文档.md b/docs/IM-Kafka接入联调测试文档.md index 62f9578..9162144 100644 --- a/docs/IM-Kafka接入联调测试文档.md +++ b/docs/IM-Kafka接入联调测试文档.md @@ -545,27 +545,27 @@ Payload： 1. 用户 A 与用户 B 建立私聊。 2. A 发送带 `x-im-client-mid` 的私聊消息。 3. 查询 Kafka `im.message.persist`。 -4. 查询租户库 `im_messages`。 +4. 查询租户私聊分表 `im_%03d_messages`。 5. A 使用同一 `x-im-client-mid` 重复发送。 预期： 1. A 立即收到服务端 ack 和 `seq_id`。 2. `seq_id` 为 BIGINT/int64。 -3. 消费成功后 `im_messages` 有一条记录。 +3. 消费成功后对应租户私聊分表有一条记录。 4. 重复发送不重复落库。 5. 消费日志 succeeded。 ### TC-KAFKA-014 群聊消息异步落库 -目标：验证群消息写入 `im_group_messages`，不混入私聊表。 +目标：验证群消息写入租户月度群消息分表，不混入私聊表。 步骤： 1. 创建群并加入至少两个成员。 2. 成员发送群消息。 -3. 查询 `im_group_messages`。 -4. 查询 `im_messages`。 +3. 查询 `im_%03d_group_messages_YYYYMM`。 +4. 查询 `im_%03d_messages`。 5. 重连客户端按 Topic seq 拉取消息。 预期： @@ -574,24 +574,27 @@ Payload： 2. `seq_id` 单调增长，类型为 BIGINT/int64。 3. 客户端补偿拉取成功。 4. 24 小时热消息可进入 Redis 热缓存。 +5. 私聊与群聊分别由 `im.message.persist`、`im.group.message.persist` 消费。 ### TC-KAFKA-015 消息消费失败重试 -目标：验证 `im.message.persist` 消费失败可重试并最终成功。 +目标：验证私聊和群聊批量落库失败后均按“逐条降级、重试 3 次、死信后提交 offset”处理。 步骤： 1. 临时制造租户库写入失败，如锁表或停止库连接。 -2. 发送消息。 -3. 恢复数据库。 -4. 等待重试。 +2. 在批次窗口内连续发送多条私聊和群聊消息。 +3. 恢复数据库，验证逐条重试可成功落库。 +4. 再次持续制造单条失败直至超过重试次数。 +5. 查询 `im_kafka_consume_log`、`im_kafka_dead_letters` 和 `im.dead.letter`。 预期： 1. `im_kafka_consume_log.retry_count` 增加。 -2. 未超过重试上限时不进入死信。 -3. 数据库恢复后消息最终落库。 -4. 状态变为 succeeded。 +2. 批量失败后降级为单条写入，每条最多执行 4 次（首次加 3 次重试）。 +3. 未超过重试上限时不进入死信，数据库恢复后消息最终落库并标记 succeeded。 +4. 持续失败的单条消息写入死信表并投递死信 Topic。 +5. 只有批内全部消息达到 succeeded/dead 终态后才提交 offset，不能跳过失败消息。 ### TC-KAFKA-016 文件上传完成事件 diff --git a/server/db/adapter.go b/server/db/adapter.go index 27f2dab..3e9d29b 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -260,6 +260,11 @@ type TenantRoutedAdapter interface { TenantAdapterFor(t.TenantID) (TenantBusinessAdapter, TenantAuthAdapter, error) } +type TenantMessageBatchAdapter interface { + TenantMessageSaveBatch(t.TenantID, []*t.Message) error + TenantGroupMessageSaveBatch(t.TenantID, []*t.Message) error +} + // TenantHTTPInfrastructureAdapter is the optional durable infrastructure used // by Kafka dead-letter handling. type TenantHTTPInfrastructureAdapter interface { diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index cb46a77..7c3b0a9 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -2249,6 +2249,41 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error { return err } +func (a *adapter) TenantMessageSaveBatch(id t.TenantID, messages []*t.Message) error { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil { + return err + } + if len(messages) == 0 { + return t.ErrMalformed + } + if tenantAdp != a { + return tenantAdp.TenantMessageSaveBatch(id, messages) + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + values := make([]string, 0, len(messages)) + args := make([]any, 0, len(messages)*13) + for _, msg := range messages { + if msg == nil || msg.TenantID != id || msg.Topic == "" || msg.SeqId <= 0 || msg.Uid().IsZero() { + return t.ErrMalformed + } + values = append(values, "(?,?,?,?,?,?,?,?,?,?,?,?,?)") + args = append(args, id, store.DecodeUid(msg.Uid()), msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic, + decodeUID(t.ParseUid(msg.From)), sql.NullString{String: msg.ClientMsgID, Valid: msg.ClientMsgID != ""}, + messageVersion(msg.Version), msg.Head, common.ToJSON(msg.Content), msg.ContentNonce, msg.KeyVersion) + } + query := fmt.Sprintf("INSERT INTO %s(tenant_id,id,createdat,updatedat,seqid,topic,`from`,client_msg_id,version,head,content,content_nonce,key_version) VALUES %s", + tenantMessagesTable(id), strings.Join(values, ",")) + _, err = a.db.ExecContext(ctx, query, args...) + if isDupe(err) { + return t.ErrDuplicate + } + return err +} + func (a *adapter) TenantGroupMessageSave(id t.TenantID, msg *t.Message) error { tenantAdp, err := a.routedTenantAdapter(id) if err != nil || msg == nil || msg.TenantID != id || msg.Topic == "" || msg.SeqId <= 0 { @@ -2315,6 +2350,69 @@ func (a *adapter) TenantGroupMessageSave(id t.TenantID, msg *t.Message) error { return tx.Commit() } +func (a *adapter) TenantGroupMessageSaveBatch(id t.TenantID, messages []*t.Message) error { + tenantAdp, err := a.routedTenantAdapter(id) + if err != nil { + return err + } + if len(messages) == 0 { + return t.ErrMalformed + } + if tenantAdp != a { + return tenantAdp.TenantGroupMessageSaveBatch(id, messages) + } + if err = a.ensureCurrentGroupMessageShardWindow(); err != nil { + return err + } + ctx, cancel := a.getContext() + if cancel != nil { + defer cancel() + } + month := messages[0].ShardMonth + table, ok := groupMessageShardTable(id, month) + if !ok { + return t.ErrMalformed + } + groupIDs := make(map[string]int64) + for _, msg := range messages { + if msg == nil || msg.TenantID != id || msg.Topic == "" || msg.SeqId <= 0 || msg.Uid().IsZero() || msg.ShardMonth != month { + return t.ErrMalformed + } + if _, found := groupIDs[msg.Topic]; !found { + groupIDs[msg.Topic], err = a.groupIDByTopic(ctx, id, msg.Topic) + if err != nil { + return err + } + } + } + var shardStatus string + if err = a.db.QueryRowContext(ctx, "SELECT status FROM im_group_message_shards WHERE tenant_id=? AND shard_month=? AND table_name=?", id, month, table).Scan(&shardStatus); err != nil { + return err + } + if shardStatus != "writable" && shardStatus != "sealing" { + return fmt.Errorf("mysql: group message shard %s is %s", month, shardStatus) + } + values := make([]string, 0, len(messages)) + args := make([]any, 0, len(messages)*20) + for _, msg := range messages { + contentJSON := common.ToJSON(msg.Content) + values = append(values, "(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)") + args = append(args, id, store.DecodeUid(msg.Uid()), msg.Topic, groupIDs[msg.Topic], msg.SeqId, + decodeUID(t.ParseUid(msg.From)), sql.NullString{String: msg.ClientMsgID, Valid: msg.ClientMsgID != ""}, + messageHeadType(msg.Head), "normal", contentJSON, msg.Head, messageHeadJSON(msg.Head, "mentioned_user_ids"), + messageHeadBool(msg.Head, "mention_all"), 0, len(contentJSON), messageVersion(msg.Version), msg.CreatedAt, + msg.UpdatedAt, msg.ContentNonce, msg.KeyVersion) + } + query := fmt.Sprintf(`INSERT INTO %s + (tenant_id,message_id,topic,group_id,seqid,from_user,client_msg_id,msg_type,status,content,head,mentioned_user_ids,mention_all, + file_count,content_size,version,created_at,updated_at,content_nonce,key_version) VALUES %s`, table, strings.Join(values, ",")) + _, err = a.db.ExecContext(ctx, query, args...) + if isDupe(err) { + return t.ErrDuplicate + } + return err +} + func (a *adapter) TenantMessageGetByClientID(id t.TenantID, from t.Uid, clientMsgID string) (*t.Message, error) { if !a.tenantScopedDB { tenantAdp, err := a.routedTenantAdapter(id) diff --git a/server/kafka_admin_commands.go b/server/kafka_admin_commands.go index 0c2ee95..003fa23 100644 --- a/server/kafka_admin_commands.go +++ b/server/kafka_admin_commands.go @@ -141,7 +141,6 @@ func handleKafkaAdminCommand(ctx context.Context, envelope kafkaEnvelope, _ kafk return nil } if kafkaCommandReasonRetryable(reason) { - sleepKafkaRetryBackoff(failedRecord.RetryCount) return execErr } if err := publishKafkaAdminCommandResult(ctx, envelope, payload.CommandID, "failed", reason, execErr.Error(), result); err != nil { @@ -608,7 +607,7 @@ func markKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, envelope if err != nil { return err } - return publishKafkaDeadLetter(ctx, "dead_letter", envelope.EventID, kafkaEnvelope{ + if err = publishKafkaDeadLetter(ctx, "dead_letter", envelope.EventID, kafkaEnvelope{ EventID: deadID, TraceID: envelope.TraceID, TenantCode: envelope.TenantCode, @@ -621,7 +620,12 @@ func markKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, envelope OccurredAt: now, ParentEventID: envelope.EventID, Payload: deadPayload, - }) + }); err != nil { + return err + } + // Persist the terminal state before the source consumer commits its offset. + // The dead-letter consumer repeats this idempotent upsert for recovery and auditing. + return store.HTTPInfrastructure.MarkKafkaConsumeDead(tenantID, dead, now) } func shouldDeadLetterKafkaCommand(reason string, retryCount int) bool { @@ -643,21 +647,6 @@ func kafkaCommandReasonRetryable(reason string) bool { } } -func sleepKafkaRetryBackoff(retryCount int) { - if kafkaBus == nil || retryCount <= 0 { - return - } - backoffs := kafkaBus.config.RetryBackoffMS - index := retryCount - 1 - if index >= len(backoffs) { - index = len(backoffs) - 1 - } - delay := time.Duration(backoffs[index]) * time.Millisecond - if delay > 0 { - time.Sleep(delay) - } -} - func parseKafkaCommandUser(raw string) types.Uid { raw = strings.TrimSpace(raw) if raw == "" { diff --git a/server/kafka_cache_invalidate.go b/server/kafka_cache_invalidate.go index d6abb76..98e9e83 100644 --- a/server/kafka_cache_invalidate.go +++ b/server/kafka_cache_invalidate.go @@ -70,7 +70,6 @@ func handleKafkaCacheInvalidateEvent(ctx context.Context, envelope kafkaEnvelope if failed.RetryCount >= kafkaBus.config.MaxRetryCount { return markKafkaDeadLetter(ctx, tenantID, envelope, failed, reason, err.Error()) } - sleepKafkaRetryBackoff(failed.RetryCount) return err } return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, consumerGroup, time.Now().UTC().Round(time.Millisecond)) diff --git a/server/kafka_config_events.go b/server/kafka_config_events.go index ac9b77c..69c1dbd 100644 --- a/server/kafka_config_events.go +++ b/server/kafka_config_events.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "strings" "time" @@ -12,6 +13,8 @@ import ( "github.com/tinode/chat/server/store/types" ) +var errKafkaConsumeInProgress = errors.New("kafka event is already being processed") + const ( configEventTenantUpdated = "tenant.updated" configEventTenantStatusChanged = "tenant.status.changed" @@ -66,7 +69,6 @@ func handleKafkaConfigEvent(ctx context.Context, envelope kafkaEnvelope, _ kafka if failed.RetryCount >= kafkaBus.config.MaxRetryCount { return markKafkaDeadLetter(ctx, tenantID, envelope, failed, reason, err.Error()) } - sleepKafkaRetryBackoff(failed.RetryCount) return err } return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, consumerGroup, time.Now().UTC().Round(time.Millisecond)) @@ -155,7 +157,7 @@ func beginKafkaConsumeForTopicWithGroup(tenantID types.TenantID, envelope kafkaE consumerGroup = kafkaBus.config.ConsumerGroup } deadline := time.Now().UTC().Add(5 * time.Minute).Round(time.Millisecond) - return store.HTTPInfrastructure.BeginKafkaConsume(tenantID, &types.KafkaConsumeRecord{ + record, claimed, err := store.HTTPInfrastructure.BeginKafkaConsume(tenantID, &types.KafkaConsumeRecord{ EventID: envelope.EventID, ConsumerGroup: consumerGroup, SourceTopic: kafkaTopic(topicName), @@ -163,4 +165,11 @@ func beginKafkaConsumeForTopicWithGroup(tenantID types.TenantID, envelope kafkaE BizType: envelope.BizType, BizID: envelope.BizID, }, deadline) + if err != nil || claimed || record == nil { + return record, claimed, err + } + if record.Status != types.KafkaConsumeSucceeded && record.Status != types.KafkaConsumeDead { + return record, false, errKafkaConsumeInProgress + } + return record, false, nil } diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go index fc11970..9259bc1 100644 --- a/server/kafka_message_persist.go +++ b/server/kafka_message_persist.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "strings" "time" "github.com/segmentio/kafka-go" @@ -24,8 +25,12 @@ type kafkaMessagePersistPayload struct { var publishKafkaMessagePersist = publishKafkaEnvelope func startKafkaMessagePersistConsumer() { - startKafkaConsumer("message_persist", handleKafkaMessagePersist) - startKafkaConsumer("group_message_persist", handleKafkaGroupMessagePersist) + startKafkaBatchConsumer("message_persist", func(ctx context.Context, messages []kafka.Message) error { + return handleKafkaMessagePersistBatch(ctx, messages, "message_persist") + }) + startKafkaBatchConsumer("group_message_persist", func(ctx context.Context, messages []kafka.Message) error { + return handleKafkaMessagePersistBatch(ctx, messages, "group_message_persist") + }) } func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) (*types.MessageSaveResult, error) { @@ -132,70 +137,155 @@ func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, ms return publishKafkaMessagePersist(ctx, kafkaMessagePersistTopicName(msg.Topic), msg.Topic, envelope) } -func handleKafkaMessagePersist(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error { - return handleKafkaMessagePersistFromTopic(ctx, envelope, "message_persist") +type kafkaMessagePersistItem struct { + envelope kafkaEnvelope + tenantID types.TenantID + payload kafkaMessagePersistPayload } -func handleKafkaGroupMessagePersist(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error { - return handleKafkaMessagePersistFromTopic(ctx, envelope, "group_message_persist") +func handleKafkaMessagePersistBatch(ctx context.Context, messages []kafka.Message, sourceTopicName string) error { + groups := make(map[string][]*kafkaMessagePersistItem) + for _, message := range messages { + item, terminal, err := prepareKafkaMessagePersistItem(ctx, message, sourceTopicName) + if err != nil { + return err + } + if terminal { + continue + } + key := fmt.Sprintf("%d", item.tenantID) + if item.payload.MessageKind == "group" { + key += ":" + item.payload.Message.ShardMonth + } + groups[key] = append(groups[key], item) + } + for _, items := range groups { + batch := make([]store.PreparedMessageBatchItem, 0, len(items)) + for _, item := range items { + batch = append(batch, store.PreparedMessageBatchItem{Message: &item.payload.Message, ReadBySender: item.payload.ReadBySender}) + } + if err := store.PersistPreparedMessageBatch(items[0].tenantID, batch); err == nil { + for _, item := range items { + if err = finishKafkaMessagePersist(ctx, item); err != nil { + if err = persistKafkaMessageIndividually(ctx, item); err != nil { + return err + } + } + } + continue + } else { + logs.Warn.Printf("kafka message batch persist failed, degrading to singles: tenant=%d kind=%s shard=%s size=%d err=%v", + items[0].tenantID, items[0].payload.MessageKind, items[0].payload.Message.ShardMonth, len(items), err) + } + for _, item := range items { + if err := persistKafkaMessageIndividually(ctx, item); err != nil { + return err + } + } + } + return nil } -func handleKafkaMessagePersistFromTopic(ctx context.Context, envelope kafkaEnvelope, sourceTopicName string) error { +func prepareKafkaMessagePersistItem(ctx context.Context, message kafka.Message, sourceTopicName string) (*kafkaMessagePersistItem, bool, error) { + var envelope kafkaEnvelope + if err := json.Unmarshal(message.Value, &envelope); err != nil { + logs.Err.Printf("kafka message malformed envelope skipped: partition=%d offset=%d err=%v", message.Partition, message.Offset, err) + return nil, true, nil + } tenantID := envelope.TenantID var err error if tenantID.IsZero() { tenantID, err = kafkaTenantID(envelope.TenantCode) if err != nil { - return err + return nil, false, err } } consume, claimed, err := beginKafkaConsumeForTopic(tenantID, envelope, sourceTopicName) - if err != nil || !claimed { - return err + if err != nil { + return nil, false, err } - var payload kafkaMessagePersistPayload - if err = json.Unmarshal(envelope.Payload, &payload); err != nil { - return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error()) + if !claimed { + return nil, true, nil } - if payload.Message.TenantID == types.ZeroTenantID { - payload.Message.TenantID = tenantID + item := &kafkaMessagePersistItem{envelope: envelope, tenantID: tenantID} + if err = json.Unmarshal(envelope.Payload, &item.payload); err != nil { + return nil, true, markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error()) } - if payload.Message.ShardMonth == "" { - payload.Message.ShardMonth = payload.ShardMonth + if item.payload.Message.TenantID == types.ZeroTenantID { + item.payload.Message.TenantID = tenantID } - if payload.Message.TenantID != tenantID { - return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "TENANT_MISMATCH", "message tenant does not match envelope tenant") + if item.payload.Message.ShardMonth == "" { + item.payload.Message.ShardMonth = item.payload.ShardMonth } - if err = validateKafkaMessageKind(payload); err != nil { - return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error()) + if item.payload.Message.TenantID != tenantID { + return nil, true, markKafkaDeadLetter(ctx, tenantID, envelope, consume, "TENANT_MISMATCH", "message tenant does not match envelope tenant") } - if _, err = store.PersistPreparedMessage(tenantID, &payload.Message, nil, payload.ReadBySender); err != nil && !errors.Is(err, types.ErrDuplicate) { - failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, - kafkaCommandErrorReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond)) - if markErr != nil { - return markErr + if err = validateKafkaMessageKind(item.payload); err != nil { + return nil, true, markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error()) + } + return item, false, nil +} + +func persistKafkaMessageIndividually(ctx context.Context, item *kafkaMessagePersistItem) error { + operation := func() error { + err := store.PersistPreparedMessageBatch(item.tenantID, []store.PreparedMessageBatchItem{{ + Message: &item.payload.Message, ReadBySender: item.payload.ReadBySender, + }}) + if errors.Is(err, types.ErrUnsupported) { + _, err = store.PersistPreparedMessage(item.tenantID, &item.payload.Message, nil, item.payload.ReadBySender) } - if failed != nil && failed.RetryCount >= kafkaBus.config.MaxRetryCount { - return markKafkaDeadLetter(ctx, tenantID, envelope, failed, kafkaCommandErrorReason(err), err.Error()) - } - sleepKafkaRetryBackoff(failed.RetryCount) - return err - } - if len(payload.AttachmentURLs) > 0 { - if err = enqueueKafkaFileAttachmentsBind(ctx, tenantID, &payload.Message, payload.AttachmentURLs); err != nil { - failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, - "KAFKA_FILE_LIFECYCLE_FAILED", err.Error(), time.Now().UTC().Round(time.Millisecond)) - if markErr != nil { - return markErr - } - if failed != nil && failed.RetryCount >= kafkaBus.config.MaxRetryCount { - return markKafkaDeadLetter(ctx, tenantID, envelope, failed, "KAFKA_FILE_LIFECYCLE_FAILED", err.Error()) - } - sleepKafkaRetryBackoff(failed.RetryCount) + if err != nil && !errors.Is(err, types.ErrDuplicate) { return err } + return finishKafkaMessagePersist(ctx, item) } - return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond)) + markFailed := func(err error) (*types.KafkaConsumeRecord, error) { + 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 { + return markKafkaDeadLetter(ctx, item.tenantID, item.envelope, failed, kafkaMessagePersistFailureReason(err), err.Error()) + } + return runKafkaMessageSingleRetries(ctx, kafkaBus.config.MessageRetries, operation, markFailed, markDead, waitKafkaRetry) +} + +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 { + 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 { + return err + } + if attempt <= retryCount { + if err = wait(ctx, kafkaRetryBackoff(attempt)); err != nil { + return err + } + } + } + return markDead(failed, lastErr) +} + +func finishKafkaMessagePersist(ctx context.Context, item *kafkaMessagePersistItem) error { + if len(item.payload.AttachmentURLs) > 0 { + if err := enqueueKafkaFileAttachmentsBind(ctx, item.tenantID, &item.payload.Message, item.payload.AttachmentURLs); err != nil { + return fmt.Errorf("file lifecycle enqueue failed: %w", err) + } + } + return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(item.tenantID, item.envelope.EventID, + kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond)) +} + +func kafkaMessagePersistFailureReason(err error) string { + if err != nil && strings.Contains(err.Error(), "file lifecycle") { + return "KAFKA_FILE_LIFECYCLE_FAILED" + } + return kafkaCommandErrorReason(err) } func kafkaMessagePersistTopicName(topic string) string { diff --git a/server/kafka_runtime.go b/server/kafka_runtime.go index aa924a0..769e9d1 100644 --- a/server/kafka_runtime.go +++ b/server/kafka_runtime.go @@ -51,6 +51,9 @@ type kafkaConfig struct { SASLPassword string `json:"sasl_password"` MaxRetryCount int `json:"max_retry_count"` RetryBackoffMS []int `json:"retry_backoff_ms"` + MessageBatchSize int `json:"message_batch_size"` + MessageBatchWait int `json:"message_batch_wait_ms"` + MessageRetries int `json:"message_single_retry_count"` DeadRetentionSec int `json:"dead_letter_retention_seconds"` } @@ -180,6 +183,17 @@ func normalizeKafkaConfig(cfg *kafkaConfig) { if len(cfg.RetryBackoffMS) == 0 { cfg.RetryBackoffMS = []int{10000, 30000, 60000, 300000, 900000} } + if cfg.MessageBatchSize <= 0 { + cfg.MessageBatchSize = 100 + } else if cfg.MessageBatchSize > 500 { + cfg.MessageBatchSize = 500 + } + if cfg.MessageBatchWait <= 0 { + cfg.MessageBatchWait = 10 + } + if cfg.MessageRetries <= 0 { + cfg.MessageRetries = 3 + } if cfg.DeadRetentionSec <= 0 { cfg.DeadRetentionSec = 90 * 24 * 3600 } @@ -276,6 +290,11 @@ func publishKafkaEnvelope(ctx context.Context, topicName string, key string, env type kafkaMessageHandler func(context.Context, kafkaEnvelope, kafka.Message) error +type kafkaMessageCommitter func(context.Context, kafka.Message) error +type kafkaRetryWaiter func(context.Context, time.Duration) error +type kafkaBatchHandler func(context.Context, []kafka.Message) error +type kafkaBatchCommitter func(context.Context, ...kafka.Message) error + func startKafkaConsumer(topicName string, handler kafkaMessageHandler) { startKafkaConsumerWithGroup(topicName, "", handler) } @@ -327,24 +346,157 @@ func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessage logs.Err.Println("Kafka fetch failed:", err) continue } - var envelope kafkaEnvelope - if err = json.Unmarshal(msg.Value, &envelope); err != nil { - logs.Err.Println("Kafka envelope decode failed:", err) - _ = reader.CommitMessages(context.Background(), msg) - continue + commit := func(ctx context.Context, message kafka.Message) error { + return reader.CommitMessages(ctx, message) } - if err = handler(kafkaBus.ctx, envelope, msg); err != nil { - logs.Err.Println("Kafka handler failed:", envelope.EventID, envelope.EventType, err) - continue - } - if err = reader.CommitMessages(context.Background(), msg); err != nil { - logs.Err.Println("Kafka commit failed:", err) + if err = consumeKafkaMessage(kafkaBus.ctx, msg, handler, commit, waitKafkaRetry); err != nil { + if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) { + return + } + logs.Err.Printf("Kafka message stopped without commit: topic=%s partition=%d offset=%d err=%v", msg.Topic, msg.Partition, msg.Offset, err) } } }() logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID) } +func startKafkaBatchConsumer(topicName string, handler kafkaBatchHandler) { + if kafkaBus == nil || handler == nil { + return + } + topic := kafkaBus.topics[topicName] + if topic == "" { + topic = topicName + } + groupID := kafkaBus.config.ConsumerGroup + "." + topicName + reader := kafka.NewReader(kafka.ReaderConfig{ + Brokers: kafkaBus.config.Brokers, GroupID: groupID, Topic: topic, StartOffset: kafka.FirstOffset, + MinBytes: 1, MaxBytes: 10e6, MaxWait: time.Second, Dialer: kafkaBus.dialer(), + }) + kafkaBus.wg.Add(1) + go func() { + defer kafkaBus.wg.Done() + defer reader.Close() + for { + messages, err := fetchKafkaMessageBatch(kafkaBus.ctx, reader, kafkaBus.config.MessageBatchSize, + time.Duration(kafkaBus.config.MessageBatchWait)*time.Millisecond) + if err != nil { + if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) { + return + } + logs.Err.Println("Kafka batch fetch failed:", err) + continue + } + if err = consumeKafkaMessageBatch(kafkaBus.ctx, messages, handler, reader.CommitMessages, waitKafkaRetry); err != nil { + if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) { + return + } + logs.Err.Printf("Kafka batch stopped without commit: topic=%s size=%d err=%v", topic, len(messages), err) + } + } + }() + logs.Info.Printf("Kafka batch consumer started: topic=%s group=%s size=%d wait_ms=%d", topic, groupID, kafkaBus.config.MessageBatchSize, kafkaBus.config.MessageBatchWait) +} + +func fetchKafkaMessageBatch(ctx context.Context, reader *kafka.Reader, maxSize int, maxWait time.Duration) ([]kafka.Message, error) { + first, err := reader.FetchMessage(ctx) + if err != nil { + return nil, err + } + messages := []kafka.Message{first} + deadline := time.Now().Add(maxWait) + for len(messages) < maxSize { + remaining := time.Until(deadline) + if remaining <= 0 { + break + } + fetchCtx, cancel := context.WithTimeout(ctx, remaining) + message, fetchErr := reader.FetchMessage(fetchCtx) + cancel() + if errors.Is(fetchErr, context.DeadlineExceeded) && ctx.Err() == nil { + break + } + if fetchErr != nil { + return nil, fetchErr + } + messages = append(messages, message) + } + return messages, nil +} + +func consumeKafkaMessageBatch(ctx context.Context, messages []kafka.Message, handler kafkaBatchHandler, commit kafkaBatchCommitter, wait kafkaRetryWaiter) error { + for attempt := 1; ; attempt++ { + if err := handler(ctx, messages); err == nil { + return commitKafkaMessageBatch(ctx, messages, commit, wait) + } + if err := wait(ctx, kafkaRetryBackoff(attempt)); err != nil { + return err + } + } +} + +func consumeKafkaMessage(ctx context.Context, msg kafka.Message, handler kafkaMessageHandler, commit kafkaMessageCommitter, wait kafkaRetryWaiter) error { + var envelope kafkaEnvelope + if err := json.Unmarshal(msg.Value, &envelope); err != nil { + return commitKafkaMessage(ctx, msg, commit, wait) + } + for attempt := 1; ; attempt++ { + if err := handler(ctx, envelope, msg); err == nil { + return commitKafkaMessage(ctx, msg, commit, wait) + } + if err := wait(ctx, kafkaRetryBackoff(attempt)); err != nil { + return err + } + } +} + +func commitKafkaMessage(ctx context.Context, msg kafka.Message, commit kafkaMessageCommitter, wait kafkaRetryWaiter) error { + for attempt := 1; ; attempt++ { + if err := commit(ctx, msg); err == nil { + return nil + } + if err := wait(ctx, kafkaRetryBackoff(attempt)); err != nil { + return err + } + } +} + +func commitKafkaMessageBatch(ctx context.Context, messages []kafka.Message, commit kafkaBatchCommitter, wait kafkaRetryWaiter) error { + for attempt := 1; ; attempt++ { + if err := commit(ctx, messages...); err == nil { + return nil + } + if err := wait(ctx, kafkaRetryBackoff(attempt)); err != nil { + return err + } + } +} + +func kafkaRetryBackoff(attempt int) time.Duration { + if kafkaBus == nil || len(kafkaBus.config.RetryBackoffMS) == 0 { + return time.Second + } + if attempt < 1 { + attempt = 1 + } + index := attempt - 1 + if index >= len(kafkaBus.config.RetryBackoffMS) { + index = len(kafkaBus.config.RetryBackoffMS) - 1 + } + return time.Duration(kafkaBus.config.RetryBackoffMS[index]) * time.Millisecond +} + +func waitKafkaRetry(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + func kafkaBroadcastConsumerGroup(topicName string) string { if kafkaBus == nil { return "" diff --git a/server/kafka_runtime_batch_test.go b/server/kafka_runtime_batch_test.go new file mode 100644 index 0000000..1ac7533 --- /dev/null +++ b/server/kafka_runtime_batch_test.go @@ -0,0 +1,143 @@ +package main + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/segmentio/kafka-go" + "github.com/tinode/chat/server/store/types" +) + +func TestConsumeKafkaMessageBatchRetriesHandlerAndCommit(t *testing.T) { + handlerCalls := 0 + commitCalls := 0 + waitCalls := 0 + messages := []kafka.Message{{Offset: 1}, {Offset: 2}} + + err := consumeKafkaMessageBatch(context.Background(), messages, + func(context.Context, []kafka.Message) error { + handlerCalls++ + if handlerCalls == 1 { + return errors.New("batch failed") + } + return nil + }, + func(_ context.Context, committed ...kafka.Message) error { + commitCalls++ + if len(committed) != len(messages) { + t.Fatalf("committed %d messages, want %d", len(committed), len(messages)) + } + if commitCalls == 1 { + return errors.New("commit failed") + } + return nil + }, + func(context.Context, time.Duration) error { + waitCalls++ + return nil + }) + + if err != nil { + t.Fatalf("consumeKafkaMessageBatch() error = %v", err) + } + if handlerCalls != 2 || commitCalls != 2 || waitCalls != 2 { + t.Fatalf("calls handler=%d commit=%d wait=%d, want 2/2/2", handlerCalls, commitCalls, waitCalls) + } +} + +func TestConsumeKafkaMalformedMessageRetriesCommit(t *testing.T) { + commitCalls := 0 + waitCalls := 0 + handlerCalls := 0 + err := consumeKafkaMessage(context.Background(), kafka.Message{Value: []byte("{")}, + func(context.Context, kafkaEnvelope, kafka.Message) error { + handlerCalls++ + return nil + }, + func(context.Context, kafka.Message) error { + commitCalls++ + if commitCalls == 1 { + return errors.New("commit failed") + } + return nil + }, + func(context.Context, time.Duration) error { + waitCalls++ + return nil + }) + + if err != nil { + t.Fatalf("consumeKafkaMessage() error = %v", err) + } + if handlerCalls != 0 || commitCalls != 2 || waitCalls != 1 { + t.Fatalf("calls handler=%d commit=%d wait=%d, want 0/2/1", handlerCalls, commitCalls, waitCalls) + } +} + +func TestRunKafkaMessageSingleRetriesThenDeadLetters(t *testing.T) { + operationCalls := 0 + markFailedCalls := 0 + markDeadCalls := 0 + waitCalls := 0 + + err := runKafkaMessageSingleRetries(context.Background(), 3, + func() error { + operationCalls++ + return errors.New("write failed") + }, + func(error) (*types.KafkaConsumeRecord, error) { + 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 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) + } +} + +func TestRunKafkaMessageSingleRetriesStopsOnSuccess(t *testing.T) { + operationCalls := 0 + markFailedCalls := 0 + markDeadCalls := 0 + err := runKafkaMessageSingleRetries(context.Background(), 3, + func() error { + operationCalls++ + if operationCalls < 3 { + return errors.New("temporary failure") + } + return nil + }, + func(error) (*types.KafkaConsumeRecord, error) { + 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) + } +} diff --git a/server/kafka_workbench_events.go b/server/kafka_workbench_events.go index 04a55a5..ddbaeb6 100644 --- a/server/kafka_workbench_events.go +++ b/server/kafka_workbench_events.go @@ -56,9 +56,6 @@ func handleKafkaWorkbenchEvent(ctx context.Context, envelope kafkaEnvelope, msg if failed != nil && failed.RetryCount >= kafkaBus.config.MaxRetryCount { return markKafkaDeadLetter(ctx, event.TenantID, envelope, failed, reason, err.Error()) } - if failed != nil { - sleepKafkaRetryBackoff(failed.RetryCount) - } return err } return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(event.TenantID, event.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond)) diff --git a/server/meetingim.conf b/server/meetingim.conf index e51c15a..daa0ed3 100644 --- a/server/meetingim.conf +++ b/server/meetingim.conf @@ -128,6 +128,9 @@ "sasl_password": "gr58tNnlZOVGW3AmMbL2", "max_retry_count": 3, "retry_backoff_ms": [10000, 30000, 60000, 300000, 900000], + "message_batch_size": 100, + "message_batch_wait_ms": 10, + "message_single_retry_count": 3, "dead_letter_retention_seconds": 7776000, "topics": { "admin_command": "im.admin.command", diff --git a/server/store/store.go b/server/store/store.go index e2c5f1e..5283c57 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -3677,6 +3677,82 @@ func PersistPreparedMessage(tenantID types.TenantID, msg *types.Message, attachm return Messages.PersistPrepared(tenantID, msg, attachmentURLs, readBySender) } +type PreparedMessageBatchItem struct { + Message *types.Message + ReadBySender bool +} + +func PersistPreparedMessageBatch(tenantID types.TenantID, items []PreparedMessageBatchItem) error { + if len(items) == 0 { + return nil + } + tenantAdp, err := tenantBusinessAdapter(tenantID) + if err != nil { + return err + } + batchAdp, ok := tenantAdp.(adapter.TenantMessageBatchAdapter) + if !ok { + return types.ErrUnsupported + } + if items[0].Message == nil { + return types.ErrMalformed + } + isGroup := types.GetTopicCat(items[0].Message.Topic) == types.TopicCatGrp + month := items[0].Message.ShardMonth + stored := make([]*types.Message, 0, len(items)) + for _, item := range items { + msg := item.Message + if msg == nil || msg.TenantID != tenantID || msg.Topic == "" || msg.SeqId <= 0 || msg.Uid().IsZero() || + (types.GetTopicCat(msg.Topic) == types.TopicCatGrp) != isGroup || (isGroup && msg.ShardMonth != month) { + return types.ErrMalformed + } + clientMsgID, normalizeErr := normalizeClientMsgID(msg.Head) + if normalizeErr != nil { + return normalizeErr + } + msg.ClientMsgID = clientMsgID + storedMsg, encryptErr := encryptMessageContent(msg) + if encryptErr != nil { + return encryptErr + } + stored = append(stored, storedMsg) + } + if isGroup { + err = batchAdp.TenantGroupMessageSaveBatch(tenantID, stored) + } else { + err = batchAdp.TenantMessageSaveBatch(tenantID, stored) + } + if err != nil { + return err + } + for _, item := range items { + msg := item.Message + from := types.ParseUid(msg.From) + if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil { + return err + } + invalidateTopicCache(tenantID, msg.Topic) + if isGroup && groupMsgRuntime != nil { + if msg.ClientMsgID != "" { + _ = groupMsgRuntime.MarkDone(tenantID, msg.Topic, from, msg.ClientMsgID, msg.SeqId, msg.Uid().String()) + } + _ = groupMsgRuntime.PutShardRoute(tenantID, msg.Topic, GroupMessageShardRoute{ + Month: msg.ShardMonth, MinSeq: msg.SeqId, MaxSeq: msg.SeqId, Count: 1, + }) + _ = groupMsgRuntime.CacheHotMessage(tenantID, msg) + } + if item.ReadBySender && !from.IsZero() { + if updateErr := tenantAdp.TenantSubsUpdate(tenantID, msg.Topic, from, map[string]any{ + "RecvSeqId": msg.SeqId, "ReadSeqId": msg.SeqId, + }); updateErr == nil { + invalidateSubscriptionCache(tenantID, msg.Topic, from) + DeleteCachedUserUnreadCount(tenantID, from) + } + } + } + return nil +} + func NormalizeClientMsgID(head types.KVMap) (string, error) { return normalizeClientMsgID(head) } diff --git a/server/tinode.conf b/server/tinode.conf index e51c15a..daa0ed3 100644 --- a/server/tinode.conf +++ b/server/tinode.conf @@ -128,6 +128,9 @@ "sasl_password": "gr58tNnlZOVGW3AmMbL2", "max_retry_count": 3, "retry_backoff_ms": [10000, 30000, 60000, 300000, 900000], + "message_batch_size": 100, + "message_batch_wait_ms": 10, + "message_single_retry_count": 3, "dead_letter_retention_seconds": 7776000, "topics": { "admin_command": "im.admin.command", diff --git a/server/topic_test.go b/server/topic_test.go index 15fd07b..87272e2 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -1284,7 +1284,7 @@ func TestFilterHotMessagesForUserSkipsDeleted(t *testing.T) { } } -func TestMarkKafkaDeadLetterPublishesBeforeDBWrite(t *testing.T) { +func TestMarkKafkaDeadLetterPublishesAndWritesTerminalState(t *testing.T) { oldKafkaBus := kafkaBus oldPublisher := publishKafkaDeadLetter oldInfrastructure := store.HTTPInfrastructure @@ -1322,8 +1322,8 @@ func TestMarkKafkaDeadLetterPublishesBeforeDBWrite(t *testing.T) { if err := markKafkaDeadLetter(context.Background(), 1, original, consume, "DB_FAILED", "boom"); err != nil { t.Fatalf("markKafkaDeadLetter returned error: %v", err) } - if len(infrastructure.deadLetters) != 0 { - t.Fatalf("dead letter was written before dead_letter topic consumption: %#v", infrastructure.deadLetters) + if len(infrastructure.deadLetters) != 1 || infrastructure.deadLetters[0].DeadID == "" { + t.Fatalf("dead letter terminal state was not written: %#v", infrastructure.deadLetters) } if publishedEnvelope.EventID == "" || publishedEnvelope.TenantID != 1 { t.Fatalf("dead letter event was not published correctly: %#v", publishedEnvelope)