ErrMissingKafkaAddress=errors.New("the Kafka address has not been configured")
ErrMissingKafkaTopic=errors.New("the Kafka topic has not been configured")
ErrInconsistentConsumerLagAtStartup=errors.New("the target and max consumer lag at startup must be either both set to 0 or to a value greater than 0")
ErrInvalidMaxConsumerLagAtStartup=errors.New("the configured max consumer lag at startup must greater or equal than the configured target consumer lag")
ErrInvalidProducerMaxRecordSizeBytes=fmt.Errorf("the configured producer max record size bytes must be a value between %d and %d",minProducerRecordDataBytesLimit,maxProducerRecordDataBytesLimit)
f.IntVar(&cfg.ProducerMaxRecordSizeBytes,prefix+".producer-max-record-size-bytes",maxProducerRecordDataBytesLimit,"The maximum size of a Kafka record data that should be generated by the producer. An incoming write request larger than this size is split into multiple Kafka records. We strongly recommend to not change this setting unless for testing purposes.")
f.Int64Var(&cfg.ProducerMaxBufferedBytes,prefix+".producer-max-buffered-bytes",1024*1024*1024,"The maximum size of (uncompressed) buffered and unacknowledged produced records sent to Kafka. The produce request fails once this limit is reached. This limit is per Kafka client. 0 to disable the limit.")
consumerLagUsage:=fmt.Sprintf("Set both -%s and -%s to 0 to disable waiting for maximum consumer lag being honored at startup.",prefix+".target-consumer-lag-at-startup",prefix+".max-consumer-lag-at-startup")
f.DurationVar(&cfg.TargetConsumerLagAtStartup,prefix+".target-consumer-lag-at-startup",2*time.Second,"The best-effort maximum lag a consumer tries to achieve at startup. "+consumerLagUsage)
f.DurationVar(&cfg.MaxConsumerLagAtStartup,prefix+".max-consumer-lag-at-startup",15*time.Second,"The guaranteed maximum lag before a consumer is considered to have caught up reading from a partition at startup, becomes ACTIVE in the hash ring and passes the readiness check. "+consumerLagUsage)
// Since we issued a request for only 1 partition, we expect exactly 1 response.
ifexpected,actual:=1,len(resps);actual!=expected{
level.Error(p.logger).Log("msg",fmt.Sprintf("unexpected number of responses (expected: %d, got: %d)",expected,actual),"expected",expected,"actual",len(resps))
returnkafkaStartOffset
}
// Ensure no error occurred.
res:=resps[0]
ifres.Err!=nil{
level.Error(p.logger).Log("msg","error fetching group offset for partition","err",res.Err)
level.Warn(logger).Log("msg","partition reader failed to fetch last produced offset","err",err)
boff.Wait()
continue
}
lastProducedOffset=lastProducedOffset-1// Kafka returns the next empty offset so we must subtract 1 to get the oldest written offset.
// Ensure there are some records to consume. For example, if the partition has been inactive for a long
// time and all its records have been deleted, the partition start offset may be > 0 but there are no
// records to actually consume.
ifpartitionStartOffset>lastProducedOffset{
level.Info(logger).Log("msg","partition reader found no records to consume because partition is empty","partition_start_offset",partitionStartOffset,"last_produced_offset",lastProducedOffset)
return0,nil
}
// This message is NOT expected to be logged with a very high rate. In this log we display the last measured
// lag. If we don't have it (lag is zero value), then it will not be logged.
level.Info(loggerWithCurrentLagIfSet(logger,currLag)).Log("msg","partition reader is consuming records to honor target and max consumer lag","partition_start_offset",partitionStartOffset,"last_produced_offset",lastProducedOffset)
forboff.Ongoing(){
// Continue reading until we reached the desired offset.
iflastProducedOffset<=p.lastProcessedOffset{
break
}
records:=p.poll(ctx)
recordsChan<-records
}
ifboff.Err()!=nil{
return0,boff.ErrCause()
}
// If it took less than the max desired lag to replay the partition
// then we can stop here, otherwise we'll have to redo it.
benclive
1 year ago
committed by