mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-18 21:26:56 +00:00
Chris LuandGitHub
c438c5ef94
* 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.
70 lines
2.0 KiB
Go
70 lines
2.0 KiB
Go
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)
|
|
}
|
|
}
|