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

@@ -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

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)
+	}
+}

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)
+	}
+}