consumer.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805
  1. package sarama
  2. import (
  3. "errors"
  4. "fmt"
  5. "sync"
  6. "sync/atomic"
  7. "time"
  8. )
  9. // ConsumerMessage encapsulates a Kafka message returned by the consumer.
  10. type ConsumerMessage struct {
  11. Key, Value []byte
  12. Topic string
  13. Partition int32
  14. Offset int64
  15. Timestamp time.Time // only set if kafka is version 0.10+, inner message timestamp
  16. BlockTimestamp time.Time // only set if kafka is version 0.10+, outer (compressed) block timestamp
  17. Headers []*RecordHeader // only set if kafka is version 0.11+
  18. }
  19. // ConsumerError is what is provided to the user when an error occurs.
  20. // It wraps an error and includes the topic and partition.
  21. type ConsumerError struct {
  22. Topic string
  23. Partition int32
  24. Err error
  25. }
  26. func (ce ConsumerError) Error() string {
  27. return fmt.Sprintf("kafka: error while consuming %s/%d: %s", ce.Topic, ce.Partition, ce.Err)
  28. }
  29. // ConsumerErrors is a type that wraps a batch of errors and implements the Error interface.
  30. // It can be returned from the PartitionConsumer's Close methods to avoid the need to manually drain errors
  31. // when stopping.
  32. type ConsumerErrors []*ConsumerError
  33. func (ce ConsumerErrors) Error() string {
  34. return fmt.Sprintf("kafka: %d errors while consuming", len(ce))
  35. }
  36. // Consumer manages PartitionConsumers which process Kafka messages from brokers. You MUST call Close()
  37. // on a consumer to avoid leaks, it will not be garbage-collected automatically when it passes out of
  38. // scope.
  39. //
  40. // Sarama's Consumer type does not currently support automatic consumer-group rebalancing and offset tracking.
  41. // For Zookeeper-based tracking (Kafka 0.8.2 and earlier), the https://github.com/wvanbergen/kafka library
  42. // builds on Sarama to add this support. For Kafka-based tracking (Kafka 0.9 and later), the
  43. // https://github.com/bsm/sarama-cluster library builds on Sarama to add this support.
  44. type Consumer interface {
  45. // Topics returns the set of available topics as retrieved from the cluster
  46. // metadata. This method is the same as Client.Topics(), and is provided for
  47. // convenience.
  48. Topics() ([]string, error)
  49. // Partitions returns the sorted list of all partition IDs for the given topic.
  50. // This method is the same as Client.Partitions(), and is provided for convenience.
  51. Partitions(topic string) ([]int32, error)
  52. // ConsumePartition creates a PartitionConsumer on the given topic/partition with
  53. // the given offset. It will return an error if this Consumer is already consuming
  54. // on the given topic/partition. Offset can be a literal offset, or OffsetNewest
  55. // or OffsetOldest
  56. ConsumePartition(topic string, partition int32, offset int64) (PartitionConsumer, error)
  57. // HighWaterMarks returns the current high water marks for each topic and partition.
  58. // Consistency between partitions is not guaranteed since high water marks are updated separately.
  59. HighWaterMarks() map[string]map[int32]int64
  60. // Close shuts down the consumer. It must be called after all child
  61. // PartitionConsumers have already been closed.
  62. Close() error
  63. }
  64. type consumer struct {
  65. client Client
  66. conf *Config
  67. ownClient bool
  68. lock sync.Mutex
  69. children map[string]map[int32]*partitionConsumer
  70. brokerConsumers map[*Broker]*brokerConsumer
  71. }
  72. // NewConsumer creates a new consumer using the given broker addresses and configuration.
  73. func NewConsumer(addrs []string, config *Config) (Consumer, error) {
  74. client, err := NewClient(addrs, config)
  75. if err != nil {
  76. return nil, err
  77. }
  78. c, err := NewConsumerFromClient(client)
  79. if err != nil {
  80. return nil, err
  81. }
  82. c.(*consumer).ownClient = true
  83. return c, nil
  84. }
  85. // NewConsumerFromClient creates a new consumer using the given client. It is still
  86. // necessary to call Close() on the underlying client when shutting down this consumer.
  87. func NewConsumerFromClient(client Client) (Consumer, error) {
  88. // Check that we are not dealing with a closed Client before processing any other arguments
  89. if client.Closed() {
  90. return nil, ErrClosedClient
  91. }
  92. c := &consumer{
  93. client: client,
  94. conf: client.Config(),
  95. children: make(map[string]map[int32]*partitionConsumer),
  96. brokerConsumers: make(map[*Broker]*brokerConsumer),
  97. }
  98. return c, nil
  99. }
  100. func (c *consumer) Close() error {
  101. if c.ownClient {
  102. return c.client.Close()
  103. }
  104. return nil
  105. }
  106. func (c *consumer) Topics() ([]string, error) {
  107. return c.client.Topics()
  108. }
  109. func (c *consumer) Partitions(topic string) ([]int32, error) {
  110. return c.client.Partitions(topic)
  111. }
  112. func (c *consumer) ConsumePartition(topic string, partition int32, offset int64) (PartitionConsumer, error) {
  113. child := &partitionConsumer{
  114. consumer: c,
  115. conf: c.conf,
  116. topic: topic,
  117. partition: partition,
  118. messages: make(chan *ConsumerMessage, c.conf.ChannelBufferSize),
  119. errors: make(chan *ConsumerError, c.conf.ChannelBufferSize),
  120. feeder: make(chan *FetchResponse, 1),
  121. trigger: make(chan none, 1),
  122. dying: make(chan none),
  123. fetchSize: c.conf.Consumer.Fetch.Default,
  124. }
  125. if err := child.chooseStartingOffset(offset); err != nil {
  126. return nil, err
  127. }
  128. var leader *Broker
  129. var err error
  130. if leader, err = c.client.Leader(child.topic, child.partition); err != nil {
  131. return nil, err
  132. }
  133. if err := c.addChild(child); err != nil {
  134. return nil, err
  135. }
  136. go withRecover(child.dispatcher)
  137. go withRecover(child.responseFeeder)
  138. child.broker = c.refBrokerConsumer(leader)
  139. child.broker.input <- child
  140. return child, nil
  141. }
  142. func (c *consumer) HighWaterMarks() map[string]map[int32]int64 {
  143. c.lock.Lock()
  144. defer c.lock.Unlock()
  145. hwms := make(map[string]map[int32]int64)
  146. for topic, p := range c.children {
  147. hwm := make(map[int32]int64, len(p))
  148. for partition, pc := range p {
  149. hwm[partition] = pc.HighWaterMarkOffset()
  150. }
  151. hwms[topic] = hwm
  152. }
  153. return hwms
  154. }
  155. func (c *consumer) addChild(child *partitionConsumer) error {
  156. c.lock.Lock()
  157. defer c.lock.Unlock()
  158. topicChildren := c.children[child.topic]
  159. if topicChildren == nil {
  160. topicChildren = make(map[int32]*partitionConsumer)
  161. c.children[child.topic] = topicChildren
  162. }
  163. if topicChildren[child.partition] != nil {
  164. return ConfigurationError("That topic/partition is already being consumed")
  165. }
  166. topicChildren[child.partition] = child
  167. return nil
  168. }
  169. func (c *consumer) removeChild(child *partitionConsumer) {
  170. c.lock.Lock()
  171. defer c.lock.Unlock()
  172. delete(c.children[child.topic], child.partition)
  173. }
  174. func (c *consumer) refBrokerConsumer(broker *Broker) *brokerConsumer {
  175. c.lock.Lock()
  176. defer c.lock.Unlock()
  177. bc := c.brokerConsumers[broker]
  178. if bc == nil {
  179. bc = c.newBrokerConsumer(broker)
  180. c.brokerConsumers[broker] = bc
  181. }
  182. bc.refs++
  183. return bc
  184. }
  185. func (c *consumer) unrefBrokerConsumer(brokerWorker *brokerConsumer) {
  186. c.lock.Lock()
  187. defer c.lock.Unlock()
  188. brokerWorker.refs--
  189. if brokerWorker.refs == 0 {
  190. close(brokerWorker.input)
  191. if c.brokerConsumers[brokerWorker.broker] == brokerWorker {
  192. delete(c.brokerConsumers, brokerWorker.broker)
  193. }
  194. }
  195. }
  196. func (c *consumer) abandonBrokerConsumer(brokerWorker *brokerConsumer) {
  197. c.lock.Lock()
  198. defer c.lock.Unlock()
  199. delete(c.brokerConsumers, brokerWorker.broker)
  200. }
  201. // PartitionConsumer
  202. // PartitionConsumer processes Kafka messages from a given topic and partition. You MUST call one of Close() or
  203. // AsyncClose() on a PartitionConsumer to avoid leaks; it will not be garbage-collected automatically when it passes out
  204. // of scope.
  205. //
  206. // The simplest way of using a PartitionConsumer is to loop over its Messages channel using a for/range
  207. // loop. The PartitionConsumer will only stop itself in one case: when the offset being consumed is reported
  208. // as out of range by the brokers. In this case you should decide what you want to do (try a different offset,
  209. // notify a human, etc) and handle it appropriately. For all other error cases, it will just keep retrying.
  210. // By default, it logs these errors to sarama.Logger; if you want to be notified directly of all errors, set
  211. // your config's Consumer.Return.Errors to true and read from the Errors channel, using a select statement
  212. // or a separate goroutine. Check out the Consumer examples to see implementations of these different approaches.
  213. //
  214. // To terminate such a for/range loop while the loop is executing, call AsyncClose. This will kick off the process of
  215. // consumer tear-down & return imediately. Continue to loop, servicing the Messages channel until the teardown process
  216. // AsyncClose initiated closes it (thus terminating the for/range loop). If you've already ceased reading Messages, call
  217. // Close; this will signal the PartitionConsumer's goroutines to begin shutting down (just like AsyncClose), but will
  218. // also drain the Messages channel, harvest all errors & return them once cleanup has completed.
  219. type PartitionConsumer interface {
  220. // AsyncClose initiates a shutdown of the PartitionConsumer. This method will return immediately, after which you
  221. // should continue to service the 'Messages' and 'Errors' channels until they are empty. It is required to call this
  222. // function, or Close before a consumer object passes out of scope, as it will otherwise leak memory. You must call
  223. // this before calling Close on the underlying client.
  224. AsyncClose()
  225. // Close stops the PartitionConsumer from fetching messages. It will initiate a shutdown just like AsyncClose, drain
  226. // the Messages channel, harvest any errors & return them to the caller. Note that if you are continuing to service
  227. // the Messages channel when this function is called, you will be competing with Close for messages; consider
  228. // calling AsyncClose, instead. It is required to call this function (or AsyncClose) before a consumer object passes
  229. // out of scope, as it will otherwise leak memory. You must call this before calling Close on the underlying client.
  230. Close() error
  231. // Messages returns the read channel for the messages that are returned by
  232. // the broker.
  233. Messages() <-chan *ConsumerMessage
  234. // Errors returns a read channel of errors that occurred during consuming, if
  235. // enabled. By default, errors are logged and not returned over this channel.
  236. // If you want to implement any custom error handling, set your config's
  237. // Consumer.Return.Errors setting to true, and read from this channel.
  238. Errors() <-chan *ConsumerError
  239. // HighWaterMarkOffset returns the high water mark offset of the partition,
  240. // i.e. the offset that will be used for the next message that will be produced.
  241. // You can use this to determine how far behind the processing is.
  242. HighWaterMarkOffset() int64
  243. }
  244. type partitionConsumer struct {
  245. highWaterMarkOffset int64 // must be at the top of the struct because https://golang.org/pkg/sync/atomic/#pkg-note-BUG
  246. consumer *consumer
  247. conf *Config
  248. topic string
  249. partition int32
  250. broker *brokerConsumer
  251. messages chan *ConsumerMessage
  252. errors chan *ConsumerError
  253. feeder chan *FetchResponse
  254. trigger, dying chan none
  255. responseResult error
  256. fetchSize int32
  257. offset int64
  258. }
  259. var errTimedOut = errors.New("timed out feeding messages to the user") // not user-facing
  260. func (child *partitionConsumer) sendError(err error) {
  261. cErr := &ConsumerError{
  262. Topic: child.topic,
  263. Partition: child.partition,
  264. Err: err,
  265. }
  266. if child.conf.Consumer.Return.Errors {
  267. child.errors <- cErr
  268. } else {
  269. Logger.Println(cErr)
  270. }
  271. }
  272. func (child *partitionConsumer) dispatcher() {
  273. for range child.trigger {
  274. select {
  275. case <-child.dying:
  276. close(child.trigger)
  277. case <-time.After(child.conf.Consumer.Retry.Backoff):
  278. if child.broker != nil {
  279. child.consumer.unrefBrokerConsumer(child.broker)
  280. child.broker = nil
  281. }
  282. Logger.Printf("consumer/%s/%d finding new broker\n", child.topic, child.partition)
  283. if err := child.dispatch(); err != nil {
  284. child.sendError(err)
  285. child.trigger <- none{}
  286. }
  287. }
  288. }
  289. if child.broker != nil {
  290. child.consumer.unrefBrokerConsumer(child.broker)
  291. }
  292. child.consumer.removeChild(child)
  293. close(child.feeder)
  294. }
  295. func (child *partitionConsumer) dispatch() error {
  296. if err := child.consumer.client.RefreshMetadata(child.topic); err != nil {
  297. return err
  298. }
  299. var leader *Broker
  300. var err error
  301. if leader, err = child.consumer.client.Leader(child.topic, child.partition); err != nil {
  302. return err
  303. }
  304. child.broker = child.consumer.refBrokerConsumer(leader)
  305. child.broker.input <- child
  306. return nil
  307. }
  308. func (child *partitionConsumer) chooseStartingOffset(offset int64) error {
  309. newestOffset, err := child.consumer.client.GetOffset(child.topic, child.partition, OffsetNewest)
  310. if err != nil {
  311. return err
  312. }
  313. oldestOffset, err := child.consumer.client.GetOffset(child.topic, child.partition, OffsetOldest)
  314. if err != nil {
  315. return err
  316. }
  317. switch {
  318. case offset == OffsetNewest:
  319. child.offset = newestOffset
  320. case offset == OffsetOldest:
  321. child.offset = oldestOffset
  322. case offset >= oldestOffset && offset <= newestOffset:
  323. child.offset = offset
  324. default:
  325. return ErrOffsetOutOfRange
  326. }
  327. return nil
  328. }
  329. func (child *partitionConsumer) Messages() <-chan *ConsumerMessage {
  330. return child.messages
  331. }
  332. func (child *partitionConsumer) Errors() <-chan *ConsumerError {
  333. return child.errors
  334. }
  335. func (child *partitionConsumer) AsyncClose() {
  336. // this triggers whatever broker owns this child to abandon it and close its trigger channel, which causes
  337. // the dispatcher to exit its loop, which removes it from the consumer then closes its 'messages' and
  338. // 'errors' channel (alternatively, if the child is already at the dispatcher for some reason, that will
  339. // also just close itself)
  340. close(child.dying)
  341. }
  342. func (child *partitionConsumer) Close() error {
  343. child.AsyncClose()
  344. go withRecover(func() {
  345. for range child.messages {
  346. // drain
  347. }
  348. })
  349. var errors ConsumerErrors
  350. for err := range child.errors {
  351. errors = append(errors, err)
  352. }
  353. if len(errors) > 0 {
  354. return errors
  355. }
  356. return nil
  357. }
  358. func (child *partitionConsumer) HighWaterMarkOffset() int64 {
  359. return atomic.LoadInt64(&child.highWaterMarkOffset)
  360. }
  361. func (child *partitionConsumer) responseFeeder() {
  362. var msgs []*ConsumerMessage
  363. expiryTicker := time.NewTicker(child.conf.Consumer.MaxProcessingTime)
  364. firstAttempt := true
  365. feederLoop:
  366. for response := range child.feeder {
  367. msgs, child.responseResult = child.parseResponse(response)
  368. for i, msg := range msgs {
  369. messageSelect:
  370. select {
  371. case child.messages <- msg:
  372. firstAttempt = true
  373. case <-expiryTicker.C:
  374. if !firstAttempt {
  375. child.responseResult = errTimedOut
  376. child.broker.acks.Done()
  377. for _, msg = range msgs[i:] {
  378. child.messages <- msg
  379. }
  380. child.broker.input <- child
  381. continue feederLoop
  382. } else {
  383. // current message has not been sent, return to select
  384. // statement
  385. firstAttempt = false
  386. goto messageSelect
  387. }
  388. }
  389. }
  390. child.broker.acks.Done()
  391. }
  392. expiryTicker.Stop()
  393. close(child.messages)
  394. close(child.errors)
  395. }
  396. func (child *partitionConsumer) parseMessages(msgSet *MessageSet) ([]*ConsumerMessage, error) {
  397. var messages []*ConsumerMessage
  398. for _, msgBlock := range msgSet.Messages {
  399. for _, msg := range msgBlock.Messages() {
  400. offset := msg.Offset
  401. if msg.Msg.Version >= 1 {
  402. baseOffset := msgBlock.Offset - msgBlock.Messages()[len(msgBlock.Messages())-1].Offset
  403. offset += baseOffset
  404. }
  405. if offset < child.offset {
  406. continue
  407. }
  408. messages = append(messages, &ConsumerMessage{
  409. Topic: child.topic,
  410. Partition: child.partition,
  411. Key: msg.Msg.Key,
  412. Value: msg.Msg.Value,
  413. Offset: offset,
  414. Timestamp: msg.Msg.Timestamp,
  415. BlockTimestamp: msgBlock.Msg.Timestamp,
  416. })
  417. child.offset = offset + 1
  418. }
  419. }
  420. if len(messages) == 0 {
  421. return nil, ErrIncompleteResponse
  422. }
  423. return messages, nil
  424. }
  425. func (child *partitionConsumer) parseRecords(batch *RecordBatch) ([]*ConsumerMessage, error) {
  426. var messages []*ConsumerMessage
  427. for _, rec := range batch.Records {
  428. offset := batch.FirstOffset + rec.OffsetDelta
  429. if offset < child.offset {
  430. continue
  431. }
  432. messages = append(messages, &ConsumerMessage{
  433. Topic: child.topic,
  434. Partition: child.partition,
  435. Key: rec.Key,
  436. Value: rec.Value,
  437. Offset: offset,
  438. Timestamp: batch.FirstTimestamp.Add(rec.TimestampDelta),
  439. Headers: rec.Headers,
  440. })
  441. child.offset = offset + 1
  442. }
  443. if len(messages) == 0 {
  444. return nil, ErrIncompleteResponse
  445. }
  446. return messages, nil
  447. }
  448. func (child *partitionConsumer) parseResponse(response *FetchResponse) ([]*ConsumerMessage, error) {
  449. block := response.GetBlock(child.topic, child.partition)
  450. if block == nil {
  451. return nil, ErrIncompleteResponse
  452. }
  453. if block.Err != ErrNoError {
  454. return nil, block.Err
  455. }
  456. nRecs, err := block.numRecords()
  457. if err != nil {
  458. return nil, err
  459. }
  460. if nRecs == 0 {
  461. partialTrailingMessage, err := block.isPartial()
  462. if err != nil {
  463. return nil, err
  464. }
  465. // We got no messages. If we got a trailing one then we need to ask for more data.
  466. // Otherwise we just poll again and wait for one to be produced...
  467. if partialTrailingMessage {
  468. if child.conf.Consumer.Fetch.Max > 0 && child.fetchSize == child.conf.Consumer.Fetch.Max {
  469. // we can't ask for more data, we've hit the configured limit
  470. child.sendError(ErrMessageTooLarge)
  471. child.offset++ // skip this one so we can keep processing future messages
  472. } else {
  473. child.fetchSize *= 2
  474. if child.conf.Consumer.Fetch.Max > 0 && child.fetchSize > child.conf.Consumer.Fetch.Max {
  475. child.fetchSize = child.conf.Consumer.Fetch.Max
  476. }
  477. }
  478. }
  479. return nil, nil
  480. }
  481. // we got messages, reset our fetch size in case it was increased for a previous request
  482. child.fetchSize = child.conf.Consumer.Fetch.Default
  483. atomic.StoreInt64(&child.highWaterMarkOffset, block.HighWaterMarkOffset)
  484. messages := []*ConsumerMessage{}
  485. for _, records := range block.RecordsSet {
  486. if control, err := records.isControl(); err != nil || control {
  487. continue
  488. }
  489. switch records.recordsType {
  490. case legacyRecords:
  491. messageSetMessages, err := child.parseMessages(records.MsgSet)
  492. if err != nil {
  493. return nil, err
  494. }
  495. messages = append(messages, messageSetMessages...)
  496. case defaultRecords:
  497. recordBatchMessages, err := child.parseRecords(records.RecordBatch)
  498. if err != nil {
  499. return nil, err
  500. }
  501. messages = append(messages, recordBatchMessages...)
  502. default:
  503. return nil, fmt.Errorf("unknown records type: %v", records.recordsType)
  504. }
  505. }
  506. return messages, nil
  507. }
  508. // brokerConsumer
  509. type brokerConsumer struct {
  510. consumer *consumer
  511. broker *Broker
  512. input chan *partitionConsumer
  513. newSubscriptions chan []*partitionConsumer
  514. wait chan none
  515. subscriptions map[*partitionConsumer]none
  516. acks sync.WaitGroup
  517. refs int
  518. }
  519. func (c *consumer) newBrokerConsumer(broker *Broker) *brokerConsumer {
  520. bc := &brokerConsumer{
  521. consumer: c,
  522. broker: broker,
  523. input: make(chan *partitionConsumer),
  524. newSubscriptions: make(chan []*partitionConsumer),
  525. wait: make(chan none),
  526. subscriptions: make(map[*partitionConsumer]none),
  527. refs: 0,
  528. }
  529. go withRecover(bc.subscriptionManager)
  530. go withRecover(bc.subscriptionConsumer)
  531. return bc
  532. }
  533. func (bc *brokerConsumer) subscriptionManager() {
  534. var buffer []*partitionConsumer
  535. // The subscriptionManager constantly accepts new subscriptions on `input` (even when the main subscriptionConsumer
  536. // goroutine is in the middle of a network request) and batches it up. The main worker goroutine picks
  537. // up a batch of new subscriptions between every network request by reading from `newSubscriptions`, so we give
  538. // it nil if no new subscriptions are available. We also write to `wait` only when new subscriptions is available,
  539. // so the main goroutine can block waiting for work if it has none.
  540. for {
  541. if len(buffer) > 0 {
  542. select {
  543. case event, ok := <-bc.input:
  544. if !ok {
  545. goto done
  546. }
  547. buffer = append(buffer, event)
  548. case bc.newSubscriptions <- buffer:
  549. buffer = nil
  550. case bc.wait <- none{}:
  551. }
  552. } else {
  553. select {
  554. case event, ok := <-bc.input:
  555. if !ok {
  556. goto done
  557. }
  558. buffer = append(buffer, event)
  559. case bc.newSubscriptions <- nil:
  560. }
  561. }
  562. }
  563. done:
  564. close(bc.wait)
  565. if len(buffer) > 0 {
  566. bc.newSubscriptions <- buffer
  567. }
  568. close(bc.newSubscriptions)
  569. }
  570. func (bc *brokerConsumer) subscriptionConsumer() {
  571. <-bc.wait // wait for our first piece of work
  572. // the subscriptionConsumer ensures we will get nil right away if no new subscriptions is available
  573. for newSubscriptions := range bc.newSubscriptions {
  574. bc.updateSubscriptions(newSubscriptions)
  575. if len(bc.subscriptions) == 0 {
  576. // We're about to be shut down or we're about to receive more subscriptions.
  577. // Either way, the signal just hasn't propagated to our goroutine yet.
  578. <-bc.wait
  579. continue
  580. }
  581. response, err := bc.fetchNewMessages()
  582. if err != nil {
  583. Logger.Printf("consumer/broker/%d disconnecting due to error processing FetchRequest: %s\n", bc.broker.ID(), err)
  584. bc.abort(err)
  585. return
  586. }
  587. bc.acks.Add(len(bc.subscriptions))
  588. for child := range bc.subscriptions {
  589. child.feeder <- response
  590. }
  591. bc.acks.Wait()
  592. bc.handleResponses()
  593. }
  594. }
  595. func (bc *brokerConsumer) updateSubscriptions(newSubscriptions []*partitionConsumer) {
  596. for _, child := range newSubscriptions {
  597. bc.subscriptions[child] = none{}
  598. Logger.Printf("consumer/broker/%d added subscription to %s/%d\n", bc.broker.ID(), child.topic, child.partition)
  599. }
  600. for child := range bc.subscriptions {
  601. select {
  602. case <-child.dying:
  603. Logger.Printf("consumer/broker/%d closed dead subscription to %s/%d\n", bc.broker.ID(), child.topic, child.partition)
  604. close(child.trigger)
  605. delete(bc.subscriptions, child)
  606. default:
  607. break
  608. }
  609. }
  610. }
  611. func (bc *brokerConsumer) handleResponses() {
  612. // handles the response codes left for us by our subscriptions, and abandons ones that have been closed
  613. for child := range bc.subscriptions {
  614. result := child.responseResult
  615. child.responseResult = nil
  616. switch result {
  617. case nil:
  618. break
  619. case errTimedOut:
  620. Logger.Printf("consumer/broker/%d abandoned subscription to %s/%d because consuming was taking too long\n",
  621. bc.broker.ID(), child.topic, child.partition)
  622. delete(bc.subscriptions, child)
  623. case ErrOffsetOutOfRange:
  624. // there's no point in retrying this it will just fail the same way again
  625. // shut it down and force the user to choose what to do
  626. child.sendError(result)
  627. Logger.Printf("consumer/%s/%d shutting down because %s\n", child.topic, child.partition, result)
  628. close(child.trigger)
  629. delete(bc.subscriptions, child)
  630. case ErrUnknownTopicOrPartition, ErrNotLeaderForPartition, ErrLeaderNotAvailable, ErrReplicaNotAvailable:
  631. // not an error, but does need redispatching
  632. Logger.Printf("consumer/broker/%d abandoned subscription to %s/%d because %s\n",
  633. bc.broker.ID(), child.topic, child.partition, result)
  634. child.trigger <- none{}
  635. delete(bc.subscriptions, child)
  636. default:
  637. // dunno, tell the user and try redispatching
  638. child.sendError(result)
  639. Logger.Printf("consumer/broker/%d abandoned subscription to %s/%d because %s\n",
  640. bc.broker.ID(), child.topic, child.partition, result)
  641. child.trigger <- none{}
  642. delete(bc.subscriptions, child)
  643. }
  644. }
  645. }
  646. func (bc *brokerConsumer) abort(err error) {
  647. bc.consumer.abandonBrokerConsumer(bc)
  648. _ = bc.broker.Close() // we don't care about the error this might return, we already have one
  649. for child := range bc.subscriptions {
  650. child.sendError(err)
  651. child.trigger <- none{}
  652. }
  653. for newSubscriptions := range bc.newSubscriptions {
  654. if len(newSubscriptions) == 0 {
  655. <-bc.wait
  656. continue
  657. }
  658. for _, child := range newSubscriptions {
  659. child.sendError(err)
  660. child.trigger <- none{}
  661. }
  662. }
  663. }
  664. func (bc *brokerConsumer) fetchNewMessages() (*FetchResponse, error) {
  665. request := &FetchRequest{
  666. MinBytes: bc.consumer.conf.Consumer.Fetch.Min,
  667. MaxWaitTime: int32(bc.consumer.conf.Consumer.MaxWaitTime / time.Millisecond),
  668. }
  669. if bc.consumer.conf.Version.IsAtLeast(V0_10_0_0) {
  670. request.Version = 2
  671. }
  672. if bc.consumer.conf.Version.IsAtLeast(V0_10_1_0) {
  673. request.Version = 3
  674. request.MaxBytes = MaxResponseSize
  675. }
  676. if bc.consumer.conf.Version.IsAtLeast(V0_11_0_0) {
  677. request.Version = 4
  678. request.Isolation = ReadUncommitted // We don't support yet transactions.
  679. }
  680. for child := range bc.subscriptions {
  681. request.AddBlock(child.topic, child.partition, child.offset, child.fetchSize)
  682. }
  683. return bc.broker.Fetch(request)
  684. }