scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 17:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

[FB-1617] Commit Messages out of order (#2180)

Browse source 
* commiting partition highmark offset only; 
* protected spool access with lock
This commit is contained in:
tgruben 2022-08-08 14:41:22 -05:00 • committed by GitHub
parent 3681feeeb2
commit 7cfa516f74
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 101 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

82
idk/kafka/source.go
View file

@ -14,6 +14,7 @@ import (
"net/http"
"net/url"
"path"
"sort"
"strconv"
"strings"
"sync"
@ -44,6 +45,7 @@ type Source struct {
spoolBase uint64
spool []confluent.TopicPartition
highmarks []confluent.TopicPartition
client *confluent.Consumer
recordChannel chan recordWithError
@ -63,6 +65,7 @@ type Source struct {
quit chan struct{}
wg sync.WaitGroup
opened bool
mu sync.Mutex
}
const defaultRegistryHost = "localhost:8081"
@ -80,6 +83,7 @@ func NewSource() *Source {
recordChannel: make(chan recordWithError),
quit: make(chan struct{}),
ConfigMap: &confluent.ConfigMap{},
highmarks: make([]confluent.TopicPartition, 0),
}
src.SchemaRegistryURL = "http://" + defaultRegistryHost
@ -116,6 +120,8 @@ func (s *Source) Record() (idk.Record, error) {
// where we should pick up from... we don't want to re-read this
// message, so we add 1
msg.TopicPartition.Offset++
s.mu.Lock()
defer s.mu.Unlock()
s.spool = append(s.spool, msg.TopicPartition)
return &Record{
src: s,
@ -151,12 +157,15 @@ func (s *Source) Schema() []idk.Field {
func (s *Source) SchemaSubject() string {
return s.schema.Subject
}
func (s *Source) SchemaSchema() string {
return s.schema.Schema
}
func (s *Source) SchemaVersion() int {
return s.schema.Version
}
func (s *Source) SchemaID() int {
return s.schema.ID
}
@ -195,19 +204,44 @@ func (r *Record) StreamOffset() (string, uint64) {
var _ idk.OffsetStreamRecord = &Record{}
func (r *Record) Commit(ctx context.Context) error {
r.src.mu.Lock()
defer r.src.mu.Unlock()
idx, base := r.idx, r.src.spoolBase
if idx < base {
return errors.New("cannot commit a record that has already been committed")
}
section, remaining := r.src.spool[:idx-base], r.src.spool[idx-base:]
committedOffsets, err := r.src.CommitMessages(section)
extra := calOffsetDiff(section, committedOffsets)
if r.src.Verbose {
for _, o := range section {
r.src.Log.Debugf("raw p: %v o: %v", o.Partition, o.Offset)
}
}
sort.Slice(section, func(i, j int) bool {
if section[i].Partition != section[j].Partition {
return section[i].Partition < section[j].Partition
}
return section[i].Offset > section[j].Offset
})
p := int32(-1)
r.src.highmarks = r.src.highmarks[:0]
// sort by increasing partition, decreasing offset
for _, x := range section {
if p != x.Partition {
r.src.highmarks = append(r.src.highmarks, x)
}
p = x.Partition
}
committedOffsets, err := r.src.CommitMessages(r.src.highmarks)
if err != nil {
return errors.Wrap(err, "failed to commit messages")
}
if len(extra) > 0 {
remaining = append(remaining, extra...)
idx = idx - uint64(len(extra))
if r.src.Verbose {
r.src.Log.Debugf("COMMIT")
for _, o := range committedOffsets {
r.src.Log.Debugf("p: %v o: %v", o.Partition, o.Offset)
}
}
r.src.spool = remaining
@ -223,6 +257,7 @@ func (r *Record) Data() []interface{} {
func calOffsetDiff(section, committed []confluent.TopicPartition) []confluent.TopicPartition {
return section[len(committed):]
}
func (s *Source) CommitMessages(recs []confluent.TopicPartition) ([]confluent.TopicPartition, error) {
return s.client.CommitOffsets(recs)
}
@ -327,12 +362,18 @@ func (c *Source) generator() {
defer func() {
close(c.recordChannel)
c.wg.Done()
if c.Verbose {
c.Log.Debugf("generator")
}
}()
for {
select {
case <-c.quit:
if c.Verbose {
c.Log.Debugf("source quit")
}
return
default:
ev := c.client.Poll(100)
@ -346,6 +387,9 @@ func (c *Source) generator() {
case confluent.AssignedPartitions:
err := c.client.Assign(e.Partitions)
if err != nil {
if c.Verbose {
c.Log.Debugf("quit AssignedParitions")
}
return
}
// If we received an `RevokedPartitions` event, we need to revoke this
@ -354,6 +398,9 @@ func (c *Source) generator() {
case confluent.RevokedPartitions:
err := c.client.Unassign()
if err != nil {
if c.Verbose {
c.Log.Debugf("quit RevokeParkitions")
}
return
}
@ -365,6 +412,9 @@ func (c *Source) generator() {
select {
case c.recordChannel <- msg:
case <-c.quit:
if c.Verbose {
c.Log.Debugf("source quit Error")
}
return
}
@ -381,9 +431,22 @@ func (c *Source) generator() {
select {
case c.recordChannel <- msg:
case <-c.quit:
if c.Verbose {
c.Log.Debugf("source quit Message")
}
return
}
case confluent.OffsetsCommitted:
c.Log.Debugf("commited %s", e)
if c.Verbose {
for _, r := range e.Offsets {
c.Log.Debugf("p: %v o: %v", r.Partition, r.Offset)
}
}
default:
if c.Verbose {
c.Log.Debugf("ignored %#v", ev)
}
continue // consumer doesn't care about all event types (e.g. OffsetsCommitted)
}
}
@ -393,7 +456,7 @@ func (c *Source) generator() {
// Close closes the underlying kafka consumer.
func (s *Source) Close() error {
if s.client != nil {
if s.opened { //only close opened sources
if s.opened { // only close opened sources
s.quit <- struct{}{}
s.wg.Wait()
err := s.client.Close()
@ -784,6 +847,7 @@ func cacheConfigProp(p propper) (*idk.CacheConfig, error) {
return &idk.CacheConfig{CacheType: pilosaclient.CacheType(cacheType), CacheSize: cacheSize}, nil
}
func intProp(p propper, s string) (int64, error) {
ival, ok := p.Prop(s)
if !ok {
@ -815,8 +879,10 @@ type propper interface {
Prop(string) (interface{}, bool)
}
var notFound = errors.New("prop not found")
var wrongType = errors.New("val is wrong type")
var (
notFound = errors.New("prop not found")
wrongType = errors.New("val is wrong type")
)
// avroUnionToPDKField takes an avro SchemaField with a Union type,
// and reduces it to a SchemaField with the type of one of the Types

34
idk/kafka_sasl/source.go
View file

@ -5,6 +5,7 @@ import (
"encoding/json"
"io"
"io/ioutil"
"sort"
"strconv"
"sync"
"time"
@ -35,6 +36,7 @@ type Source struct {
spoolBase uint64
spool []confluent.TopicPartition
highmarks []confluent.TopicPartition
client *confluent.Consumer
recordChannel chan recordWithError
ConfigMap *confluent.ConfigMap
@ -43,6 +45,7 @@ type Source struct {
quit chan struct{}
wg sync.WaitGroup
opened bool
mu sync.Mutex
}
// NewSource gets a new Source
@ -86,6 +89,8 @@ func (s *Source) Record() (idk.Record, error) {
// where we should pick up from... we don't want to re-read this
// message, so we add 1
msg.TopicPartition.Offset++
s.mu.Lock()
defer s.mu.Unlock()
s.spool = append(s.spool, msg.TopicPartition)
return &Record{
src: s,
@ -124,6 +129,7 @@ func (s *Source) decodeMessage(buf []byte) ([]interface{}, error) {
func (s *Source) Schema() []idk.Field {
return s.schema
}
func (s *Source) CommitMessages(recs []confluent.TopicPartition) ([]confluent.TopicPartition, error) {
return s.client.CommitOffsets(recs)
}
@ -144,20 +150,34 @@ func (r *Record) StreamOffset() (string, uint64) {
var _ idk.OffsetStreamRecord = &Record{}
func (r *Record) Commit(ctx context.Context) error {
r.src.mu.Lock()
defer r.src.mu.Unlock()
idx, base := r.idx, r.src.spoolBase
if idx < base {
return errors.New("cannot commit a record that has already been committed")
}
section, remaining := r.src.spool[:idx-base], r.src.spool[idx-base:]
committedOffsets, err := r.src.CommitMessages(section)
extra := calOffsetDiff(section, committedOffsets)
// sort by increasing partition, decreasing offset
sort.Slice(section, func(i, j int) bool {
if section[i].Partition != section[j].Partition {
return section[i].Partition < section[j].Partition
}
return section[i].Offset > section[j].Offset
})
// calculate the high marks
p := int32(-1)
r.src.highmarks = r.src.highmarks[:0]
for _, x := range section {
if p != x.Partition {
r.src.highmarks = append(r.src.highmarks, x)
}
p = x.Partition
}
_, err := r.src.CommitMessages(r.src.highmarks)
if err != nil {
return errors.Wrap(err, "failed to commit messages")
}
if len(extra) > 0 {
remaining = append(remaining, extra...)
idx = idx - uint64(len(extra))
}
r.src.spool = remaining
r.src.spoolBase = idx
@ -306,7 +326,7 @@ func (c *Source) generator() {
// Close closes the underlying kafka consumer.
func (s *Source) Close() error {
if s.client != nil {
if s.opened { //only close opened sources
if s.opened { // only close opened sources
s.quit <- struct{}{}
s.wg.Wait()
err := s.client.Close()