mirror of
https://gitee.com/johng/gf
synced 2026-07-04 04:52:48 +08:00
thirdparty package kafka updated to date
This commit is contained in:
John
@ -25,27 +25,49 @@ type offsetManager struct {
|
||||
client Client
|
||||
conf *Config
|
||||
group string
|
||||
ticker *time.Ticker
|
||||
|
||||
lock sync.Mutex
|
||||
poms map[string]map[int32]*partitionOffsetManager
|
||||
boms map[*Broker]*brokerOffsetManager
|
||||
memberID string
|
||||
generation int32
|
||||
|
||||
broker *Broker
|
||||
brokerLock sync.RWMutex
|
||||
|
||||
poms map[string]map[int32]*partitionOffsetManager
|
||||
pomsLock sync.RWMutex
|
||||
|
||||
closeOnce sync.Once
|
||||
closing chan none
|
||||
closed chan none
|
||||
}
|
||||
|
||||
// NewOffsetManagerFromClient creates a new OffsetManager from the given client.
|
||||
// It is still necessary to call Close() on the underlying client when finished with the partition manager.
|
||||
func NewOffsetManagerFromClient(group string, client Client) (OffsetManager, error) {
|
||||
return newOffsetManagerFromClient(group, "", GroupGenerationUndefined, client)
|
||||
}
|
||||
|
||||
func newOffsetManagerFromClient(group, memberID string, generation int32, client Client) (*offsetManager, error) {
|
||||
// Check that we are not dealing with a closed Client before processing any other arguments
|
||||
if client.Closed() {
|
||||
return nil, ErrClosedClient
|
||||
}
|
||||
|
||||
conf := client.Config()
|
||||
om := &offsetManager{
|
||||
client: client,
|
||||
conf: client.Config(),
|
||||
conf: conf,
|
||||
group: group,
|
||||
ticker: time.NewTicker(conf.Consumer.Offsets.CommitInterval),
|
||||
poms: make(map[string]map[int32]*partitionOffsetManager),
|
||||
boms: make(map[*Broker]*brokerOffsetManager),
|
||||
|
||||
memberID: memberID,
|
||||
generation: generation,
|
||||
|
||||
closing: make(chan none),
|
||||
closed: make(chan none),
|
||||
}
|
||||
go withRecover(om.mainLoop)
|
||||
|
||||
return om, nil
|
||||
}
|
||||
@ -56,8 +78,8 @@ func (om *offsetManager) ManagePartition(topic string, partition int32) (Partiti
|
||||
return nil, err
|
||||
}
|
||||
|
||||
om.lock.Lock()
|
||||
defer om.lock.Unlock()
|
||||
om.pomsLock.Lock()
|
||||
defer om.pomsLock.Unlock()
|
||||
|
||||
topicManagers := om.poms[topic]
|
||||
if topicManagers == nil {
|
||||
@ -74,53 +96,307 @@ func (om *offsetManager) ManagePartition(topic string, partition int32) (Partiti
|
||||
}
|
||||
|
||||
func (om *offsetManager) Close() error {
|
||||
om.closeOnce.Do(func() {
|
||||
// exit the mainLoop
|
||||
close(om.closing)
|
||||
<-om.closed
|
||||
|
||||
// mark all POMs as closed
|
||||
om.asyncClosePOMs()
|
||||
|
||||
// flush one last time
|
||||
for attempt := 0; attempt <= om.conf.Consumer.Offsets.Retry.Max; attempt++ {
|
||||
om.flushToBroker()
|
||||
if om.releasePOMs(false) == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
om.releasePOMs(true)
|
||||
om.brokerLock.Lock()
|
||||
om.broker = nil
|
||||
om.brokerLock.Unlock()
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (om *offsetManager) refBrokerOffsetManager(broker *Broker) *brokerOffsetManager {
|
||||
om.lock.Lock()
|
||||
defer om.lock.Unlock()
|
||||
|
||||
bom := om.boms[broker]
|
||||
if bom == nil {
|
||||
bom = om.newBrokerOffsetManager(broker)
|
||||
om.boms[broker] = bom
|
||||
func (om *offsetManager) fetchInitialOffset(topic string, partition int32, retries int) (int64, string, error) {
|
||||
broker, err := om.coordinator()
|
||||
if err != nil {
|
||||
if retries <= 0 {
|
||||
return 0, "", err
|
||||
}
|
||||
return om.fetchInitialOffset(topic, partition, retries-1)
|
||||
}
|
||||
|
||||
bom.refs++
|
||||
req := new(OffsetFetchRequest)
|
||||
req.Version = 1
|
||||
req.ConsumerGroup = om.group
|
||||
req.AddPartition(topic, partition)
|
||||
|
||||
return bom
|
||||
resp, err := broker.FetchOffset(req)
|
||||
if err != nil {
|
||||
if retries <= 0 {
|
||||
return 0, "", err
|
||||
}
|
||||
om.releaseCoordinator(broker)
|
||||
return om.fetchInitialOffset(topic, partition, retries-1)
|
||||
}
|
||||
|
||||
block := resp.GetBlock(topic, partition)
|
||||
if block == nil {
|
||||
return 0, "", ErrIncompleteResponse
|
||||
}
|
||||
|
||||
switch block.Err {
|
||||
case ErrNoError:
|
||||
return block.Offset, block.Metadata, nil
|
||||
case ErrNotCoordinatorForConsumer:
|
||||
if retries <= 0 {
|
||||
return 0, "", block.Err
|
||||
}
|
||||
om.releaseCoordinator(broker)
|
||||
return om.fetchInitialOffset(topic, partition, retries-1)
|
||||
case ErrOffsetsLoadInProgress:
|
||||
if retries <= 0 {
|
||||
return 0, "", block.Err
|
||||
}
|
||||
select {
|
||||
case <-om.closing:
|
||||
return 0, "", block.Err
|
||||
case <-time.After(om.conf.Metadata.Retry.Backoff):
|
||||
}
|
||||
return om.fetchInitialOffset(topic, partition, retries-1)
|
||||
default:
|
||||
return 0, "", block.Err
|
||||
}
|
||||
}
|
||||
|
||||
func (om *offsetManager) unrefBrokerOffsetManager(bom *brokerOffsetManager) {
|
||||
om.lock.Lock()
|
||||
defer om.lock.Unlock()
|
||||
func (om *offsetManager) coordinator() (*Broker, error) {
|
||||
om.brokerLock.RLock()
|
||||
broker := om.broker
|
||||
om.brokerLock.RUnlock()
|
||||
|
||||
bom.refs--
|
||||
if broker != nil {
|
||||
return broker, nil
|
||||
}
|
||||
|
||||
if bom.refs == 0 {
|
||||
close(bom.updateSubscriptions)
|
||||
if om.boms[bom.broker] == bom {
|
||||
delete(om.boms, bom.broker)
|
||||
om.brokerLock.Lock()
|
||||
defer om.brokerLock.Unlock()
|
||||
|
||||
if broker := om.broker; broker != nil {
|
||||
return broker, nil
|
||||
}
|
||||
|
||||
if err := om.client.RefreshCoordinator(om.group); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
broker, err := om.client.Coordinator(om.group)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
om.broker = broker
|
||||
return broker, nil
|
||||
}
|
||||
|
||||
func (om *offsetManager) releaseCoordinator(b *Broker) {
|
||||
om.brokerLock.Lock()
|
||||
if om.broker == b {
|
||||
om.broker = nil
|
||||
}
|
||||
om.brokerLock.Unlock()
|
||||
}
|
||||
|
||||
func (om *offsetManager) mainLoop() {
|
||||
defer om.ticker.Stop()
|
||||
defer close(om.closed)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-om.ticker.C:
|
||||
om.flushToBroker()
|
||||
om.releasePOMs(false)
|
||||
case <-om.closing:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (om *offsetManager) abandonBroker(bom *brokerOffsetManager) {
|
||||
om.lock.Lock()
|
||||
defer om.lock.Unlock()
|
||||
func (om *offsetManager) flushToBroker() {
|
||||
req := om.constructRequest()
|
||||
if req == nil {
|
||||
return
|
||||
}
|
||||
|
||||
delete(om.boms, bom.broker)
|
||||
broker, err := om.coordinator()
|
||||
if err != nil {
|
||||
om.handleError(err)
|
||||
return
|
||||
}
|
||||
|
||||
resp, err := broker.CommitOffset(req)
|
||||
if err != nil {
|
||||
om.handleError(err)
|
||||
om.releaseCoordinator(broker)
|
||||
_ = broker.Close()
|
||||
return
|
||||
}
|
||||
|
||||
om.handleResponse(broker, req, resp)
|
||||
}
|
||||
|
||||
func (om *offsetManager) abandonPartitionOffsetManager(pom *partitionOffsetManager) {
|
||||
om.lock.Lock()
|
||||
defer om.lock.Unlock()
|
||||
func (om *offsetManager) constructRequest() *OffsetCommitRequest {
|
||||
var r *OffsetCommitRequest
|
||||
var perPartitionTimestamp int64
|
||||
if om.conf.Consumer.Offsets.Retention == 0 {
|
||||
perPartitionTimestamp = ReceiveTime
|
||||
r = &OffsetCommitRequest{
|
||||
Version: 1,
|
||||
ConsumerGroup: om.group,
|
||||
ConsumerID: om.memberID,
|
||||
ConsumerGroupGeneration: om.generation,
|
||||
}
|
||||
} else {
|
||||
r = &OffsetCommitRequest{
|
||||
Version: 2,
|
||||
RetentionTime: int64(om.conf.Consumer.Offsets.Retention / time.Millisecond),
|
||||
ConsumerGroup: om.group,
|
||||
ConsumerID: om.memberID,
|
||||
ConsumerGroupGeneration: om.generation,
|
||||
}
|
||||
|
||||
delete(om.poms[pom.topic], pom.partition)
|
||||
if len(om.poms[pom.topic]) == 0 {
|
||||
delete(om.poms, pom.topic)
|
||||
}
|
||||
|
||||
om.pomsLock.RLock()
|
||||
defer om.pomsLock.RUnlock()
|
||||
|
||||
for _, topicManagers := range om.poms {
|
||||
for _, pom := range topicManagers {
|
||||
pom.lock.Lock()
|
||||
if pom.dirty {
|
||||
r.AddBlock(pom.topic, pom.partition, pom.offset, perPartitionTimestamp, pom.metadata)
|
||||
}
|
||||
pom.lock.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
if len(r.blocks) > 0 {
|
||||
return r
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (om *offsetManager) handleResponse(broker *Broker, req *OffsetCommitRequest, resp *OffsetCommitResponse) {
|
||||
om.pomsLock.RLock()
|
||||
defer om.pomsLock.RUnlock()
|
||||
|
||||
for _, topicManagers := range om.poms {
|
||||
for _, pom := range topicManagers {
|
||||
if req.blocks[pom.topic] == nil || req.blocks[pom.topic][pom.partition] == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
var err KError
|
||||
var ok bool
|
||||
|
||||
if resp.Errors[pom.topic] == nil {
|
||||
pom.handleError(ErrIncompleteResponse)
|
||||
continue
|
||||
}
|
||||
if err, ok = resp.Errors[pom.topic][pom.partition]; !ok {
|
||||
pom.handleError(ErrIncompleteResponse)
|
||||
continue
|
||||
}
|
||||
|
||||
switch err {
|
||||
case ErrNoError:
|
||||
block := req.blocks[pom.topic][pom.partition]
|
||||
pom.updateCommitted(block.offset, block.metadata)
|
||||
case ErrNotLeaderForPartition, ErrLeaderNotAvailable,
|
||||
ErrConsumerCoordinatorNotAvailable, ErrNotCoordinatorForConsumer:
|
||||
// not a critical error, we just need to redispatch
|
||||
om.releaseCoordinator(broker)
|
||||
case ErrOffsetMetadataTooLarge, ErrInvalidCommitOffsetSize:
|
||||
// nothing we can do about this, just tell the user and carry on
|
||||
pom.handleError(err)
|
||||
case ErrOffsetsLoadInProgress:
|
||||
// nothing wrong but we didn't commit, we'll get it next time round
|
||||
break
|
||||
case ErrUnknownTopicOrPartition:
|
||||
// let the user know *and* try redispatching - if topic-auto-create is
|
||||
// enabled, redispatching should trigger a metadata req and create the
|
||||
// topic; if not then re-dispatching won't help, but we've let the user
|
||||
// know and it shouldn't hurt either (see https://github.com/Shopify/sarama/issues/706)
|
||||
fallthrough
|
||||
default:
|
||||
// dunno, tell the user and try redispatching
|
||||
pom.handleError(err)
|
||||
om.releaseCoordinator(broker)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (om *offsetManager) handleError(err error) {
|
||||
om.pomsLock.RLock()
|
||||
defer om.pomsLock.RUnlock()
|
||||
|
||||
for _, topicManagers := range om.poms {
|
||||
for _, pom := range topicManagers {
|
||||
pom.handleError(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (om *offsetManager) asyncClosePOMs() {
|
||||
om.pomsLock.RLock()
|
||||
defer om.pomsLock.RUnlock()
|
||||
|
||||
for _, topicManagers := range om.poms {
|
||||
for _, pom := range topicManagers {
|
||||
pom.AsyncClose()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Releases/removes closed POMs once they are clean (or when forced)
|
||||
func (om *offsetManager) releasePOMs(force bool) (remaining int) {
|
||||
om.pomsLock.Lock()
|
||||
defer om.pomsLock.Unlock()
|
||||
|
||||
for topic, topicManagers := range om.poms {
|
||||
for partition, pom := range topicManagers {
|
||||
pom.lock.Lock()
|
||||
releaseDue := pom.done && (force || !pom.dirty)
|
||||
pom.lock.Unlock()
|
||||
|
||||
if releaseDue {
|
||||
pom.release()
|
||||
|
||||
delete(om.poms[topic], partition)
|
||||
if len(om.poms[topic]) == 0 {
|
||||
delete(om.poms, topic)
|
||||
}
|
||||
}
|
||||
}
|
||||
remaining += len(om.poms[topic])
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (om *offsetManager) findPOM(topic string, partition int32) *partitionOffsetManager {
|
||||
om.pomsLock.RLock()
|
||||
defer om.pomsLock.RUnlock()
|
||||
|
||||
if partitions, ok := om.poms[topic]; ok {
|
||||
if pom, ok := partitions[partition]; ok {
|
||||
return pom
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Partition Offset Manager
|
||||
@ -187,138 +463,26 @@ type partitionOffsetManager struct {
|
||||
offset int64
|
||||
metadata string
|
||||
dirty bool
|
||||
clean sync.Cond
|
||||
broker *brokerOffsetManager
|
||||
done bool
|
||||
|
||||
errors chan *ConsumerError
|
||||
rebalance chan none
|
||||
dying chan none
|
||||
releaseOnce sync.Once
|
||||
errors chan *ConsumerError
|
||||
}
|
||||
|
||||
func (om *offsetManager) newPartitionOffsetManager(topic string, partition int32) (*partitionOffsetManager, error) {
|
||||
pom := &partitionOffsetManager{
|
||||
offset, metadata, err := om.fetchInitialOffset(topic, partition, om.conf.Metadata.Retry.Max)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &partitionOffsetManager{
|
||||
parent: om,
|
||||
topic: topic,
|
||||
partition: partition,
|
||||
errors: make(chan *ConsumerError, om.conf.ChannelBufferSize),
|
||||
rebalance: make(chan none, 1),
|
||||
dying: make(chan none),
|
||||
}
|
||||
pom.clean.L = &pom.lock
|
||||
|
||||
if err := pom.selectBroker(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := pom.fetchInitialOffset(om.conf.Metadata.Retry.Max); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pom.broker.updateSubscriptions <- pom
|
||||
|
||||
go withRecover(pom.mainLoop)
|
||||
|
||||
return pom, nil
|
||||
}
|
||||
|
||||
func (pom *partitionOffsetManager) mainLoop() {
|
||||
for {
|
||||
select {
|
||||
case <-pom.rebalance:
|
||||
if err := pom.selectBroker(); err != nil {
|
||||
pom.handleError(err)
|
||||
pom.rebalance <- none{}
|
||||
} else {
|
||||
pom.broker.updateSubscriptions <- pom
|
||||
}
|
||||
case <-pom.dying:
|
||||
if pom.broker != nil {
|
||||
select {
|
||||
case <-pom.rebalance:
|
||||
case pom.broker.updateSubscriptions <- pom:
|
||||
}
|
||||
pom.parent.unrefBrokerOffsetManager(pom.broker)
|
||||
}
|
||||
pom.parent.abandonPartitionOffsetManager(pom)
|
||||
close(pom.errors)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (pom *partitionOffsetManager) selectBroker() error {
|
||||
if pom.broker != nil {
|
||||
pom.parent.unrefBrokerOffsetManager(pom.broker)
|
||||
pom.broker = nil
|
||||
}
|
||||
|
||||
var broker *Broker
|
||||
var err error
|
||||
|
||||
if err = pom.parent.client.RefreshCoordinator(pom.parent.group); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if broker, err = pom.parent.client.Coordinator(pom.parent.group); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
pom.broker = pom.parent.refBrokerOffsetManager(broker)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (pom *partitionOffsetManager) fetchInitialOffset(retries int) error {
|
||||
request := new(OffsetFetchRequest)
|
||||
request.Version = 1
|
||||
request.ConsumerGroup = pom.parent.group
|
||||
request.AddPartition(pom.topic, pom.partition)
|
||||
|
||||
response, err := pom.broker.broker.FetchOffset(request)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
block := response.GetBlock(pom.topic, pom.partition)
|
||||
if block == nil {
|
||||
return ErrIncompleteResponse
|
||||
}
|
||||
|
||||
switch block.Err {
|
||||
case ErrNoError:
|
||||
pom.offset = block.Offset
|
||||
pom.metadata = block.Metadata
|
||||
return nil
|
||||
case ErrNotCoordinatorForConsumer:
|
||||
if retries <= 0 {
|
||||
return block.Err
|
||||
}
|
||||
if err := pom.selectBroker(); err != nil {
|
||||
return err
|
||||
}
|
||||
return pom.fetchInitialOffset(retries - 1)
|
||||
case ErrOffsetsLoadInProgress:
|
||||
if retries <= 0 {
|
||||
return block.Err
|
||||
}
|
||||
time.Sleep(pom.parent.conf.Metadata.Retry.Backoff)
|
||||
return pom.fetchInitialOffset(retries - 1)
|
||||
default:
|
||||
return block.Err
|
||||
}
|
||||
}
|
||||
|
||||
func (pom *partitionOffsetManager) handleError(err error) {
|
||||
cErr := &ConsumerError{
|
||||
Topic: pom.topic,
|
||||
Partition: pom.partition,
|
||||
Err: err,
|
||||
}
|
||||
|
||||
if pom.parent.conf.Consumer.Return.Errors {
|
||||
pom.errors <- cErr
|
||||
} else {
|
||||
Logger.Println(cErr)
|
||||
}
|
||||
offset: offset,
|
||||
metadata: metadata,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (pom *partitionOffsetManager) Errors() <-chan *ConsumerError {
|
||||
@ -353,7 +517,6 @@ func (pom *partitionOffsetManager) updateCommitted(offset int64, metadata string
|
||||
|
||||
if pom.offset == offset && pom.metadata == metadata {
|
||||
pom.dirty = false
|
||||
pom.clean.Signal()
|
||||
}
|
||||
}
|
||||
|
||||
@ -369,16 +532,9 @@ func (pom *partitionOffsetManager) NextOffset() (int64, string) {
|
||||
}
|
||||
|
||||
func (pom *partitionOffsetManager) AsyncClose() {
|
||||
go func() {
|
||||
pom.lock.Lock()
|
||||
defer pom.lock.Unlock()
|
||||
|
||||
for pom.dirty {
|
||||
pom.clean.Wait()
|
||||
}
|
||||
|
||||
close(pom.dying)
|
||||
}()
|
||||
pom.lock.Lock()
|
||||
pom.done = true
|
||||
pom.lock.Unlock()
|
||||
}
|
||||
|
||||
func (pom *partitionOffsetManager) Close() error {
|
||||
@ -395,166 +551,22 @@ func (pom *partitionOffsetManager) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Broker Offset Manager
|
||||
|
||||
type brokerOffsetManager struct {
|
||||
parent *offsetManager
|
||||
broker *Broker
|
||||
timer *time.Ticker
|
||||
updateSubscriptions chan *partitionOffsetManager
|
||||
subscriptions map[*partitionOffsetManager]none
|
||||
refs int
|
||||
}
|
||||
|
||||
func (om *offsetManager) newBrokerOffsetManager(broker *Broker) *brokerOffsetManager {
|
||||
bom := &brokerOffsetManager{
|
||||
parent: om,
|
||||
broker: broker,
|
||||
timer: time.NewTicker(om.conf.Consumer.Offsets.CommitInterval),
|
||||
updateSubscriptions: make(chan *partitionOffsetManager),
|
||||
subscriptions: make(map[*partitionOffsetManager]none),
|
||||
func (pom *partitionOffsetManager) handleError(err error) {
|
||||
cErr := &ConsumerError{
|
||||
Topic: pom.topic,
|
||||
Partition: pom.partition,
|
||||
Err: err,
|
||||
}
|
||||
|
||||
go withRecover(bom.mainLoop)
|
||||
|
||||
return bom
|
||||
}
|
||||
|
||||
func (bom *brokerOffsetManager) mainLoop() {
|
||||
for {
|
||||
select {
|
||||
case <-bom.timer.C:
|
||||
if len(bom.subscriptions) > 0 {
|
||||
bom.flushToBroker()
|
||||
}
|
||||
case s, ok := <-bom.updateSubscriptions:
|
||||
if !ok {
|
||||
bom.timer.Stop()
|
||||
return
|
||||
}
|
||||
if _, ok := bom.subscriptions[s]; ok {
|
||||
delete(bom.subscriptions, s)
|
||||
} else {
|
||||
bom.subscriptions[s] = none{}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (bom *brokerOffsetManager) flushToBroker() {
|
||||
request := bom.constructRequest()
|
||||
if request == nil {
|
||||
return
|
||||
}
|
||||
|
||||
response, err := bom.broker.CommitOffset(request)
|
||||
|
||||
if err != nil {
|
||||
bom.abort(err)
|
||||
return
|
||||
}
|
||||
|
||||
for s := range bom.subscriptions {
|
||||
if request.blocks[s.topic] == nil || request.blocks[s.topic][s.partition] == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
var err KError
|
||||
var ok bool
|
||||
|
||||
if response.Errors[s.topic] == nil {
|
||||
s.handleError(ErrIncompleteResponse)
|
||||
delete(bom.subscriptions, s)
|
||||
s.rebalance <- none{}
|
||||
continue
|
||||
}
|
||||
if err, ok = response.Errors[s.topic][s.partition]; !ok {
|
||||
s.handleError(ErrIncompleteResponse)
|
||||
delete(bom.subscriptions, s)
|
||||
s.rebalance <- none{}
|
||||
continue
|
||||
}
|
||||
|
||||
switch err {
|
||||
case ErrNoError:
|
||||
block := request.blocks[s.topic][s.partition]
|
||||
s.updateCommitted(block.offset, block.metadata)
|
||||
case ErrNotLeaderForPartition, ErrLeaderNotAvailable,
|
||||
ErrConsumerCoordinatorNotAvailable, ErrNotCoordinatorForConsumer:
|
||||
// not a critical error, we just need to redispatch
|
||||
delete(bom.subscriptions, s)
|
||||
s.rebalance <- none{}
|
||||
case ErrOffsetMetadataTooLarge, ErrInvalidCommitOffsetSize:
|
||||
// nothing we can do about this, just tell the user and carry on
|
||||
s.handleError(err)
|
||||
case ErrOffsetsLoadInProgress:
|
||||
// nothing wrong but we didn't commit, we'll get it next time round
|
||||
break
|
||||
case ErrUnknownTopicOrPartition:
|
||||
// let the user know *and* try redispatching - if topic-auto-create is
|
||||
// enabled, redispatching should trigger a metadata request and create the
|
||||
// topic; if not then re-dispatching won't help, but we've let the user
|
||||
// know and it shouldn't hurt either (see https://github.com/Shopify/sarama/issues/706)
|
||||
fallthrough
|
||||
default:
|
||||
// dunno, tell the user and try redispatching
|
||||
s.handleError(err)
|
||||
delete(bom.subscriptions, s)
|
||||
s.rebalance <- none{}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (bom *brokerOffsetManager) constructRequest() *OffsetCommitRequest {
|
||||
var r *OffsetCommitRequest
|
||||
var perPartitionTimestamp int64
|
||||
if bom.parent.conf.Consumer.Offsets.Retention == 0 {
|
||||
perPartitionTimestamp = ReceiveTime
|
||||
r = &OffsetCommitRequest{
|
||||
Version: 1,
|
||||
ConsumerGroup: bom.parent.group,
|
||||
ConsumerGroupGeneration: GroupGenerationUndefined,
|
||||
}
|
||||
if pom.parent.conf.Consumer.Return.Errors {
|
||||
pom.errors <- cErr
|
||||
} else {
|
||||
r = &OffsetCommitRequest{
|
||||
Version: 2,
|
||||
RetentionTime: int64(bom.parent.conf.Consumer.Offsets.Retention / time.Millisecond),
|
||||
ConsumerGroup: bom.parent.group,
|
||||
ConsumerGroupGeneration: GroupGenerationUndefined,
|
||||
}
|
||||
|
||||
Logger.Println(cErr)
|
||||
}
|
||||
|
||||
for s := range bom.subscriptions {
|
||||
s.lock.Lock()
|
||||
if s.dirty {
|
||||
r.AddBlock(s.topic, s.partition, s.offset, perPartitionTimestamp, s.metadata)
|
||||
}
|
||||
s.lock.Unlock()
|
||||
}
|
||||
|
||||
if len(r.blocks) > 0 {
|
||||
return r
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (bom *brokerOffsetManager) abort(err error) {
|
||||
_ = bom.broker.Close() // we don't care about the error this might return, we already have one
|
||||
bom.parent.abandonBroker(bom)
|
||||
|
||||
for pom := range bom.subscriptions {
|
||||
pom.handleError(err)
|
||||
pom.rebalance <- none{}
|
||||
}
|
||||
|
||||
for s := range bom.updateSubscriptions {
|
||||
if _, ok := bom.subscriptions[s]; !ok {
|
||||
s.handleError(err)
|
||||
s.rebalance <- none{}
|
||||
}
|
||||
}
|
||||
|
||||
bom.subscriptions = make(map[*partitionOffsetManager]none)
|
||||
func (pom *partitionOffsetManager) release() {
|
||||
pom.releaseOnce.Do(func() {
|
||||
go close(pom.errors)
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user