Merge pull request #60 from WillemKauf/tombstones

Compaction verification enhancements

Author: WillemKauf

cmd/kgo-repeater/main.go

@@ -50,8 +50,9 @@ var (
 	transactionAbortRate = flag.Float64("transaction-abort-rate", 0.0, "The probability that any given transaction should abort")
 	msgsPerTransaction   = flag.Uint("msgs-per-transaction", 1, "The number of messages that should be in a given transaction")
 
-	compressionType     = flag.String("compression-type", "", "One of gzip, snappy, lz4, zstd, or 'mixed' to pick a random codec for each producer")
-	compressiblePayload = flag.Bool("compressible-payload", false, "If true, use a highly compressible payload instead of the default random payload")
+	compressionType      = flag.String("compression-type", "", "One of gzip, snappy, lz4, zstd, or 'mixed' to pick a random codec for each producer")
+	compressiblePayload  = flag.Bool("compressible-payload", false, "If true, use a highly compressible payload instead of the default random payload")
+	tombstoneProbability = flag.Float64("tombstone-probability", 0.0, "The probability (between 0.0 and 1.0) that a record produced is a tombstone record.")
 )
 
 // NewAdmin returns a franz-go admin client.
@@ -150,7 +151,7 @@ func main() {
 		// it was refactored.
 		wConfig := worker.NewWorkerConfig(
 			name, *brokers, *trace, topicsList[0], *linger, *maxBufferedRecords, *useTransactions, *compressionType, *compressiblePayload, *username, *password, *enableTls)
-		config := repeater.NewRepeaterConfig(wConfig, topicsList, *group, *keys, *payloadSize, dataInFlightPerWorker, rateLimitPerWorker)
+		config := repeater.NewRepeaterConfig(wConfig, topicsList, *group, *keys, *payloadSize, dataInFlightPerWorker, rateLimitPerWorker, *tombstoneProbability)
 		lv := repeater.NewWorker(config)
 		if *useTransactions {
 			tconfig := worker.NewTransactionSTMConfig(*transactionAbortRate, *msgsPerTransaction)
@@ -212,9 +213,9 @@ func main() {
 	go func() {
 		listenAddr := fmt.Sprintf("0.0.0.0:%d", *remotePort)
 		if err := http.ListenAndServe(listenAddr, mux); err != nil {
-			panic(fmt.Sprintf("failed to listen on %s: %v", listenAddr, err));
+			panic(fmt.Sprintf("failed to listen on %s: %v", listenAddr, err))
 		}
-	}();
+	}()
 
 	if !*remote {
 		admin, err := NewAdmin()

cmd/kgo-verifier/main.go

@@ -72,6 +72,9 @@ var (
 
 	tolerateDataLoss      = flag.Bool("tolerate-data-loss", false, "If true, tolerate data-loss events")
 	tolerateFailedProduce = flag.Bool("tolerate-failed-produce", false, "If true, tolerate and retry failed produce")
+	tombstoneProbability  = flag.Float64("tombstone-probability", 0.0, "The probability (between 0.0 and 1.0) that a record produced is a tombstone record.")
+	compacted             = flag.Bool("compacted", false, "Whether the topic to be verified is compacted or not. This will suppress warnings about offset gaps in consumed values.")
+	validateLatestValues  = flag.Bool("validate-latest-values", false, "If true, values consumed by a worker will be validated against the last produced value by a producer. This value should only be set if compaction has been allowed to fully de-duplicate the entirety of the log before consuming.")
 )
 
 func makeWorkerConfig() worker.WorkerConfig {
@@ -92,6 +95,7 @@ func makeWorkerConfig() worker.WorkerConfig {
 		TolerateDataLoss:      *tolerateDataLoss,
 		TolerateFailedProduce: *tolerateFailedProduce,
 		Continuous:            *continuous,
+		ValidateLatestValues:  *validateLatestValues,
 	}
 
 	return c
@@ -245,7 +249,7 @@ func main() {
 
 	if *pCount > 0 {
 		log.Info("Starting producer...")
-		pwc := verifier.NewProducerConfig(makeWorkerConfig(), "producer", nPartitions, *mSize, *pCount, *fakeTimestampMs, *fakeTimestampStepMs, (*produceRateLimitBps), *keySetCardinality, *msgsPerProducerId)
+		pwc := verifier.NewProducerConfig(makeWorkerConfig(), "producer", nPartitions, *mSize, *pCount, *fakeTimestampMs, *fakeTimestampStepMs, (*produceRateLimitBps), *keySetCardinality, *msgsPerProducerId, *tombstoneProbability)
 		pw := verifier.NewProducerWorker(pwc)
 
 		if *useTransactions {
@@ -261,7 +265,7 @@ func main() {
 		srw := verifier.NewSeqReadWorker(verifier.NewSeqReadConfig(
 			makeWorkerConfig(), "sequential", nPartitions, *seqConsumeCount,
 			(*consumeTputMb)*1024*1024,
-		))
+		), verifier.NewValidatorStatus(*compacted, *validateLatestValues, *topic, nPartitions))
 		workers = append(workers, &srw)
 
 		for loopState.Next() {
@@ -280,7 +284,7 @@ func main() {
 			workerCfg := verifier.NewRandomReadConfig(
 				makeWorkerConfig(), fmt.Sprintf("random-%03d", i), nPartitions, *cCount,
 			)
-			worker := verifier.NewRandomReadWorker(workerCfg)
+			worker := verifier.NewRandomReadWorker(workerCfg, verifier.NewValidatorStatus(*compacted, *validateLatestValues, *topic, nPartitions))
 			randomWorkers = append(randomWorkers, &worker)
 			workers = append(workers, &worker)
 		}
@@ -309,7 +313,7 @@ func main() {
 		grw := verifier.NewGroupReadWorker(
 			verifier.NewGroupReadConfig(
 				makeWorkerConfig(), *cgName, nPartitions, *cgReaders,
-				*seqConsumeCount, (*consumeTputMb)*1024*1024))
+				*seqConsumeCount, (*consumeTputMb)*1024*1024), verifier.NewValidatorStatus(*compacted, *validateLatestValues, *topic, nPartitions))
 		workers = append(workers, &grw)
 
 		for loopState.Next() {

pkg/worker/repeater/repeater_worker.go

@@ -52,13 +52,13 @@ type RepeaterConfig struct {
 	RateLimitBps   int
 }
 
-func NewRepeaterConfig(cfg worker.WorkerConfig, topics []string, group string, keys uint64, payloadSize uint64, dataInFlight uint64, rateLimitBps int) RepeaterConfig {
+func NewRepeaterConfig(cfg worker.WorkerConfig, topics []string, group string, keys uint64, payloadSize uint64, dataInFlight uint64, rateLimitBps int, tombstoneProbability float64) RepeaterConfig {
 	return RepeaterConfig{
 		workerCfg:      cfg,
 		Topics:         topics,
 		Group:          group,
 		KeySpace:       worker.KeySpace{UniqueCount: keys},
-		ValueGenerator: worker.ValueGenerator{PayloadSize: payloadSize, Compressible: cfg.CompressiblePayload},
+		ValueGenerator: worker.ValueGenerator{PayloadSize: payloadSize, Compressible: cfg.CompressiblePayload, TombstoneProbability: tombstoneProbability},
 		DataInFlight:   dataInFlight,
 		RateLimitBps:   rateLimitBps,
 	}

pkg/worker/verifier/group_read_worker.go

@@ -49,10 +49,10 @@ type GroupReadWorker struct {
 	Status GroupWorkerStatus
 }
 
-func NewGroupReadWorker(cfg GroupReadConfig) GroupReadWorker {
+func NewGroupReadWorker(cfg GroupReadConfig, validatorStatus ValidatorStatus) GroupReadWorker {
 	return GroupReadWorker{
 		config: cfg,
-		Status: GroupWorkerStatus{Topic: cfg.workerCfg.Topic},
+		Status: GroupWorkerStatus{Topic: cfg.workerCfg.Topic, Validator: validatorStatus},
 	}
 }
 
@@ -240,6 +240,10 @@ func (grw *GroupReadWorker) consumerGroupReadInner(
 	defer client.Close()
 
 	validRanges := LoadTopicOffsetRanges(grw.config.workerCfg.Topic, grw.config.nPartitions)
+	var latestValuesProduced LatestValueMap
+	if grw.Status.Validator.expectFullyCompacted {
+		latestValuesProduced = LoadLatestValues(grw.config.workerCfg.Topic, grw.config.nPartitions)
+	}
 
 	for {
 		fetches := client.PollFetches(ctx)
@@ -271,7 +275,7 @@ func (grw *GroupReadWorker) consumerGroupReadInner(
 			log.Debugf(
 				"fiber %v: Consumer group read %s/%d o=%d...",
 				fiberId, grw.config.workerCfg.Topic, r.Partition, r.Offset)
-			grw.Status.Validator.ValidateRecord(r, &validRanges)
+			grw.Status.Validator.ValidateRecord(r, &validRanges, &latestValuesProduced)
 			// Will cancel the context if we have read everything
 			cgOffsets.AddRecord(ctx, r)
 		})

pkg/worker/verifier/latest_value_map.go

@@ -0,0 +1,92 @@
+package verifier
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+
+	"github.com/redpanda-data/kgo-verifier/pkg/util"
+	log "github.com/sirupsen/logrus"
+)
+
+type LatestValueMap struct {
+	topic               string
+	LatestKvByPartition []map[string]string
+}
+
+func (lvm *LatestValueMap) Get(partition int32, key string) (value string, exists bool) {
+	if partition < 0 || partition >= int32(len(lvm.LatestKvByPartition)) {
+		log.Panicf("Partition %d out of bounds for latestValueMap of size %d", partition, len(lvm.LatestKvByPartition))
+	}
+	value, exists = lvm.LatestKvByPartition[partition][key]
+	return
+}
+
+func (lvm *LatestValueMap) Insert(partition int32, key string, value string) {
+	lvm.LatestKvByPartition[partition][key] = value
+}
+
+func latestValueMapFile(topic string) string {
+	return fmt.Sprintf("latest_value_%s.json", topic)
+}
+
+func (lvm *LatestValueMap) Store() error {
+	log.Infof("LatestValueMap::Storing %s", latestValueMapFile(lvm.topic))
+
+	data, err := json.Marshal(lvm)
+	if err != nil {
+		return err
+	}
+
+	tmp_file, err := ioutil.TempFile("./", "latest_value_*.tmp")
+	if err != nil {
+		return err
+	}
+
+	_, err = tmp_file.Write(data)
+	if err != nil {
+		return err
+	}
+
+	err = os.Rename(tmp_file.Name(), latestValueMapFile(lvm.topic))
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func LoadLatestValues(topic string, nPartitions int32) LatestValueMap {
+	data, err := ioutil.ReadFile(latestValueMapFile(topic))
+	if err != nil {
+		util.Die("Can't read topic latest value map: %v", err)
+	}
+
+	var lvm LatestValueMap
+	if len(data) > 0 {
+		err = json.Unmarshal(data, &lvm)
+		util.Chk(err, "Bad JSON %v", err)
+	}
+
+	if int32(len(lvm.LatestKvByPartition)) > nPartitions {
+		util.Die("More partitions in latest_value_map file than in topic!")
+	} else if len(lvm.LatestKvByPartition) < int(nPartitions) {
+		// Creating new partitions is allowed
+		blanks := make([]map[string]string, nPartitions-int32(len(lvm.LatestKvByPartition)))
+		lvm.LatestKvByPartition = append(lvm.LatestKvByPartition, blanks...)
+	}
+	log.Infof("Successfully read latest value map")
+	return lvm
+}
+
+func NewLatestValueMap(topic string, nPartitions int32) LatestValueMap {
+	maps := make([]map[string]string, nPartitions)
+	for i := range maps {
+		maps[i] = make(map[string]string)
+	}
+	return LatestValueMap{
+		topic:               topic,
+		LatestKvByPartition: maps,
+	}
+}

pkg/worker/verifier/offset_ranges.go

@@ -31,6 +31,14 @@ func LoadTopicOffsetRanges(topic string, nPartitions int32) TopicOffsetRanges {
 			tors.PartitionRanges = append(tors.PartitionRanges, blanks...)
 		}
 
+		if int32(len(tors.LastConsumableOffsets)) > nPartitions {
+			util.Die("More partitions in valid_offsets file than in topic!")
+		} else if len(tors.LastConsumableOffsets) < int(nPartitions) {
+			// Creating new partitions is allowed
+			blanks := make([]int64, nPartitions-int32(len(tors.LastConsumableOffsets)))
+			tors.LastConsumableOffsets = append(tors.LastConsumableOffsets, blanks...)
+		}
+
 		return tors
 	}
 }
@@ -105,6 +113,15 @@ func (ors *OffsetRanges) Contains(o int64) bool {
 type TopicOffsetRanges struct {
 	topic           string
 	PartitionRanges []OffsetRanges
+
+	// In the case that the topic being consumed from had tombstones produced,
+	// the high watermark may be given by a tombstone record that has been removed.
+	// In trying to consume until this point, readers will become stuck polling for
+	// new records.
+	AdjustConsumableOffsets bool
+	// Persist the last consumable offset here to adjust the offset we attempt to read
+	// up to in the read workers.
+	LastConsumableOffsets []int64
 }
 
 func (tors *TopicOffsetRanges) Insert(p int32, o int64) {
@@ -115,6 +132,10 @@ func (tors *TopicOffsetRanges) Contains(p int32, o int64) bool {
 	return tors.PartitionRanges[p].Contains(o)
 }
 
+func (tors *TopicOffsetRanges) SetLastConsumableOffset(p int32, o int64) {
+	tors.LastConsumableOffsets[p] = o
+}
+
 func topicOffsetRangeFile(topic string) string {
 	return fmt.Sprintf("valid_offsets_%s.json", topic)
 }
@@ -153,8 +174,12 @@ func NewTopicOffsetRanges(topic string, nPartitions int32) TopicOffsetRanges {
 	for _, or := range prs {
 		or.Ranges = make([]OffsetRange, 0)
 	}
+	lcos := make([]int64, nPartitions)
+
 	return TopicOffsetRanges{
-		topic:           topic,
-		PartitionRanges: prs,
+		topic:                   topic,
+		PartitionRanges:         prs,
+		AdjustConsumableOffsets: false,
+		LastConsumableOffsets:   lcos,
 	}
 }