From c438c5ef946bde014eb4d31f7442521eec885a7c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 24 Jul 2026 10:43:48 -0700 Subject: [PATCH] filer.replicate: acknowledge notifications after the sink write, not on receipt (#10427) * filer.replicate: commit the kafka offset after replicating, not on receipt The partition consumer committed the offset as soon as it handed the message to the channel, so a sink write that failed was logged and the message was already behind the committed offset -- never redelivered, permanently missing from the sink. Commit in onSuccessFn instead, and hold the committed offset behind the oldest offset that failed to replicate so a restart redelivers from there. * filer.replicate: delete the sqs message after replicating, not on receipt ReceiveMessage deleted the message before the replicator had a chance to run, so a failed sink write dropped it for good. Move the delete into onSuccessFn and leave the message in the queue otherwise, letting the visibility timeout redeliver it. --- weed/replication/sub/notification_aws_sqs.go | 21 ++++-- weed/replication/sub/notification_kafka.go | 37 +++++++++- .../sub/notification_kafka_test.go | 69 +++++++++++++++++++ 3 files changed, 117 insertions(+), 10 deletions(-) create mode 100644 weed/replication/sub/notification_kafka_test.go diff --git a/weed/replication/sub/notification_aws_sqs.go b/weed/replication/sub/notification_aws_sqs.go index 5eb42c2aa..978b36bb2 100644 --- a/weed/replication/sub/notification_aws_sqs.go +++ b/weed/replication/sub/notification_aws_sqs.go @@ -103,14 +103,21 @@ func (k *AwsSqsInput) ReceiveMessage() (key string, message *filer_pb.EventNotif err = fmt.Errorf("unmarshal message from sqs %s: %w", k.queueUrl, err) return } - // delete the message - _, err = k.svc.DeleteMessage(&sqs.DeleteMessageInput{ - QueueUrl: &k.queueUrl, - ReceiptHandle: result.Messages[0].ReceiptHandle, - }) - if err != nil { - glog.V(1).Infof("delete message from sqs %s: %v", k.queueUrl, err) + // Delete only once the message has been replicated. Deleting on receipt + // drops the message for good when the sink write fails; leaving it in the + // queue lets the visibility timeout redeliver it. + receiptHandle := result.Messages[0].ReceiptHandle + onSuccessFn = func() { + if _, deleteErr := k.svc.DeleteMessage(&sqs.DeleteMessageInput{ + QueueUrl: &k.queueUrl, + ReceiptHandle: receiptHandle, + }); deleteErr != nil { + glog.V(1).Infof("delete message from sqs %s: %v", k.queueUrl, deleteErr) + } + } + onFailureFn = func() { + glog.V(1).Infof("keeping message %s in sqs %s for redelivery", key, k.queueUrl) } return diff --git a/weed/replication/sub/notification_kafka.go b/weed/replication/sub/notification_kafka.go index e5af4d84f..31a8e6892 100644 --- a/weed/replication/sub/notification_kafka.go +++ b/weed/replication/sub/notification_kafka.go @@ -23,6 +23,7 @@ type KafkaInput struct { topic string consumer sarama.Consumer messageChan chan *sarama.ConsumerMessage + progress *KafkaProgress } func (k *KafkaInput) GetName() string { @@ -81,6 +82,8 @@ func (k *KafkaInput) initialize(hosts []string, topic string, offsetFile string, progress.lastSaveTime = time.Now() progress.offsetFile = offsetFile progress.offsetSaveIntervalSeconds = offsetSaveIntervalSeconds + progress.failedOffsets = make(map[int32]int64) + k.progress = progress for _, partition := range partitions { offset, found := progress.PartitionOffsets[partition] @@ -100,9 +103,6 @@ func (k *KafkaInput) initialize(hosts []string, topic string, offsetFile string, fmt.Println(err) case msg := <-partitionConsumer.Messages(): k.messageChan <- msg - if err := progress.setOffset(msg.Partition, msg.Offset); err != nil { - glog.Warningf("set kafka offset: %v", err) - } } } }() @@ -115,6 +115,17 @@ func (k *KafkaInput) ReceiveMessage() (key string, message *filer_pb.EventNotifi msg := <-k.messageChan + // Commit only once the message has been replicated. Committing on receipt + // leaves nothing to redeliver when the sink write fails. + onSuccessFn = func() { + if err := k.progress.setOffset(msg.Partition, msg.Offset); err != nil { + glog.Warningf("set kafka offset: %v", err) + } + } + onFailureFn = func() { + k.progress.markFailed(msg.Partition, msg.Offset) + } + key = string(msg.Key) message = &filer_pb.EventNotification{} err = proto.Unmarshal(msg.Value, message) @@ -128,6 +139,10 @@ type KafkaProgress struct { offsetFile string lastSaveTime time.Time offsetSaveIntervalSeconds int + // failedOffsets is the oldest offset that failed to replicate in each + // partition. Nothing at or past it is ever committed, so a restart + // redelivers from the failure instead of resuming after it. + failedOffsets map[int32]int64 sync.Mutex } @@ -164,9 +179,25 @@ func (progress *KafkaProgress) setOffset(partition int32, offset int64) error { progress.Lock() defer progress.Unlock() + if failedOffset, found := progress.failedOffsets[partition]; found && offset >= failedOffset { + return nil + } + progress.PartitionOffsets[partition] = offset if int(time.Now().Sub(progress.lastSaveTime).Seconds()) > progress.offsetSaveIntervalSeconds { return progress.saveProgress() } return nil } + +// markFailed records an offset that could not be replicated, holding the +// committed offset for that partition behind it. +func (progress *KafkaProgress) markFailed(partition int32, offset int64) { + progress.Lock() + defer progress.Unlock() + + if failedOffset, found := progress.failedOffsets[partition]; !found || offset < failedOffset { + progress.failedOffsets[partition] = offset + glog.Errorf("replicate kafka %s partition %d offset %d failed; holding the committed offset before it", progress.Topic, partition, offset) + } +} diff --git a/weed/replication/sub/notification_kafka_test.go b/weed/replication/sub/notification_kafka_test.go new file mode 100644 index 000000000..d4cbb3987 --- /dev/null +++ b/weed/replication/sub/notification_kafka_test.go @@ -0,0 +1,69 @@ +package sub + +import ( + "testing" + "time" +) + +func newTestProgress() *KafkaProgress { + return &KafkaProgress{ + Topic: "test", + PartitionOffsets: make(map[int32]int64), + failedOffsets: make(map[int32]int64), + lastSaveTime: time.Now(), + // large enough that no test call reaches saveProgress and touches disk + offsetSaveIntervalSeconds: 3600, + } +} + +// TestKafkaProgressHoldsOffsetAtFailure verifies that a message that failed to +// replicate keeps the committed offset behind it, so a restart redelivers it +// rather than resuming after it. +func TestKafkaProgressHoldsOffsetAtFailure(t *testing.T) { + progress := newTestProgress() + + if err := progress.setOffset(0, 10); err != nil { + t.Fatalf("setOffset: %v", err) + } + if got := progress.PartitionOffsets[0]; got != 10 { + t.Fatalf("offset = %d after a replicated message, want 10", got) + } + + progress.markFailed(0, 11) + + if err := progress.setOffset(0, 12); err != nil { + t.Fatalf("setOffset: %v", err) + } + if got := progress.PartitionOffsets[0]; got != 10 { + t.Fatalf("offset = %d after a later success, want it held at 10", got) + } + + // a failure in one partition must not stall the others + if err := progress.setOffset(1, 5); err != nil { + t.Fatalf("setOffset: %v", err) + } + if got := progress.PartitionOffsets[1]; got != 5 { + t.Fatalf("offset = %d in an unaffected partition, want 5", got) + } +} + +// TestKafkaProgressKeepsOldestFailure verifies that the hold point is the +// oldest failed offset, not the most recent one. +func TestKafkaProgressKeepsOldestFailure(t *testing.T) { + progress := newTestProgress() + + progress.markFailed(0, 20) + progress.markFailed(0, 11) + progress.markFailed(0, 30) + + if got := progress.failedOffsets[0]; got != 11 { + t.Fatalf("held at offset %d, want the oldest failure 11", got) + } + + if err := progress.setOffset(0, 10); err != nil { + t.Fatalf("setOffset: %v", err) + } + if got := progress.PartitionOffsets[0]; got != 10 { + t.Fatalf("offset = %d, want an offset before the failure to still commit", got) + } +}