Skip to content

Commit 963015e

Browse files
eapacheeapache
committed
Merge pull request #368 from Shopify/flush-retry-recovery
producer: bugfix for aggregators getting stuck
2 parents e8ecee7 + a381b36 commit 963015e

File tree

3 files changed

+92
-125
lines changed

3 files changed

+92
-125
lines changed

‎CHANGELOG.md

+2
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@
55
Bug Fixes:
66
- Fix the producer's internal reference counting in certain unusual scenarios
77
([#367](https://github.com/Shopify/sarama/pull/367)).
8+
- Fix a condition where the producer's internal control messages could have
9+
gotten stuck ([#368](https://github.com/Shopify/sarama/pull/368)).
810

911
#### Version 1.0.0 (2015-03-17)
1012

‎async_producer.go

+1
Original file line numberDiff line numberDiff line change
@@ -458,6 +458,7 @@ func (p *asyncProducer) messageAggregator(broker *Broker, input chan *ProducerMe
458458
bytesAccumulated += msg.byteSize()
459459

460460
if defaultFlush ||
461+
msg.flags&chaser == chaser ||
461462
(p.conf.Producer.Flush.Messages > 0 && len(buffer) >= p.conf.Producer.Flush.Messages) ||
462463
(p.conf.Producer.Flush.Bytes > 0 && bytesAccumulated >= p.conf.Producer.Flush.Bytes) {
463464
doFlush = flusher

‎async_producer_test.go

+89-125
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,22 @@ func closeProducer(t *testing.T, p AsyncProducer) {
3030
wg.Wait()
3131
}
3232

33+
func expectSuccesses(t *testing.T, p AsyncProducer, successes int) {
34+
for i := 0; i < successes; i++ {
35+
select {
36+
case msg := <-p.Errors():
37+
t.Error(msg.Err)
38+
if msg.Msg.flags != 0 {
39+
t.Error("Message had flags set")
40+
}
41+
case msg := <-p.Successes():
42+
if msg.flags != 0 {
43+
t.Error("Message had flags set")
44+
}
45+
}
46+
}
47+
}
48+
3349
func TestAsyncProducer(t *testing.T) {
3450
seedBroker := newMockBroker(t, 1)
3551
leader := newMockBroker(t, 2)
@@ -103,19 +119,7 @@ func TestAsyncProducerMultipleFlushes(t *testing.T) {
103119
for i := 0; i < 5; i++ {
104120
producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage)}
105121
}
106-
for i := 0; i < 5; i++ {
107-
select {
108-
case msg := <-producer.Errors():
109-
t.Error(msg.Err)
110-
if msg.Msg.flags != 0 {
111-
t.Error("Message had flags set")
112-
}
113-
case msg := <-producer.Successes():
114-
if msg.flags != 0 {
115-
t.Error("Message had flags set")
116-
}
117-
}
118-
}
122+
expectSuccesses(t, producer, 5)
119123
}
120124

121125
closeProducer(t, producer)
@@ -155,19 +159,7 @@ func TestAsyncProducerMultipleBrokers(t *testing.T) {
155159
for i := 0; i < 10; i++ {
156160
producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage)}
157161
}
158-
for i := 0; i < 10; i++ {
159-
select {
160-
case msg := <-producer.Errors():
161-
t.Error(msg.Err)
162-
if msg.Msg.flags != 0 {
163-
t.Error("Message had flags set")
164-
}
165-
case msg := <-producer.Successes():
166-
if msg.flags != 0 {
167-
t.Error("Message had flags set")
168-
}
169-
}
170-
}
162+
expectSuccesses(t, producer, 10)
171163

172164
closeProducer(t, producer)
173165
leader1.Close()
@@ -210,38 +202,14 @@ func TestAsyncProducerFailureRetry(t *testing.T) {
210202
prodSuccess := new(ProduceResponse)
211203
prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
212204
leader2.Returns(prodSuccess)
213-
for i := 0; i < 10; i++ {
214-
select {
215-
case msg := <-producer.Errors():
216-
t.Error(msg.Err)
217-
if msg.Msg.flags != 0 {
218-
t.Error("Message had flags set")
219-
}
220-
case msg := <-producer.Successes():
221-
if msg.flags != 0 {
222-
t.Error("Message had flags set")
223-
}
224-
}
225-
}
205+
expectSuccesses(t, producer, 10)
226206
leader1.Close()
227207

228208
for i := 0; i < 10; i++ {
229209
producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage)}
230210
}
231211
leader2.Returns(prodSuccess)
232-
for i := 0; i < 10; i++ {
233-
select {
234-
case msg := <-producer.Errors():
235-
t.Error(msg.Err)
236-
if msg.Msg.flags != 0 {
237-
t.Error("Message had flags set")
238-
}
239-
case msg := <-producer.Successes():
240-
if msg.flags != 0 {
241-
t.Error("Message had flags set")
242-
}
243-
}
244-
}
212+
expectSuccesses(t, producer, 10)
245213

246214
leader2.Close()
247215
closeProducer(t, producer)
@@ -276,19 +244,7 @@ func TestAsyncProducerBrokerBounce(t *testing.T) {
276244
prodSuccess := new(ProduceResponse)
277245
prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
278246
leader.Returns(prodSuccess)
279-
for i := 0; i < 10; i++ {
280-
select {
281-
case msg := <-producer.Errors():
282-
t.Error(msg.Err)
283-
if msg.Msg.flags != 0 {
284-
t.Error("Message had flags set")
285-
}
286-
case msg := <-producer.Successes():
287-
if msg.flags != 0 {
288-
t.Error("Message had flags set")
289-
}
290-
}
291-
}
247+
expectSuccesses(t, producer, 10)
292248
seedBroker.Close()
293249
leader.Close()
294250

@@ -331,19 +287,7 @@ func TestAsyncProducerBrokerBounceWithStaleMetadata(t *testing.T) {
331287
prodSuccess := new(ProduceResponse)
332288
prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
333289
leader2.Returns(prodSuccess)
334-
for i := 0; i < 10; i++ {
335-
select {
336-
case msg := <-producer.Errors():
337-
t.Error(msg.Err)
338-
if msg.Msg.flags != 0 {
339-
t.Error("Message had flags set")
340-
}
341-
case msg := <-producer.Successes():
342-
if msg.flags != 0 {
343-
t.Error("Message had flags set")
344-
}
345-
}
346-
}
290+
expectSuccesses(t, producer, 10)
347291
seedBroker.Close()
348292
leader2.Close()
349293

@@ -391,37 +335,13 @@ func TestAsyncProducerMultipleRetries(t *testing.T) {
391335
prodSuccess := new(ProduceResponse)
392336
prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
393337
leader2.Returns(prodSuccess)
394-
for i := 0; i < 10; i++ {
395-
select {
396-
case msg := <-producer.Errors():
397-
t.Error(msg.Err)
398-
if msg.Msg.flags != 0 {
399-
t.Error("Message had flags set")
400-
}
401-
case msg := <-producer.Successes():
402-
if msg.flags != 0 {
403-
t.Error("Message had flags set")
404-
}
405-
}
406-
}
338+
expectSuccesses(t, producer, 10)
407339

408340
for i := 0; i < 10; i++ {
409341
producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage)}
410342
}
411343
leader2.Returns(prodSuccess)
412-
for i := 0; i < 10; i++ {
413-
select {
414-
case msg := <-producer.Errors():
415-
t.Error(msg.Err)
416-
if msg.Msg.flags != 0 {
417-
t.Error("Message had flags set")
418-
}
419-
case msg := <-producer.Successes():
420-
if msg.flags != 0 {
421-
t.Error("Message had flags set")
422-
}
423-
}
424-
}
344+
expectSuccesses(t, producer, 10)
425345

426346
seedBroker.Close()
427347
leader1.Close()
@@ -479,13 +399,7 @@ func TestAsyncProducerOutOfRetries(t *testing.T) {
479399
prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
480400
leader.Returns(prodSuccess)
481401

482-
for i := 0; i < 10; i++ {
483-
select {
484-
case msg := <-producer.Errors():
485-
t.Error(msg.Err)
486-
case <-producer.Successes():
487-
}
488-
}
402+
expectSuccesses(t, producer, 10)
489403

490404
leader.Close()
491405
seedBroker.Close()
@@ -518,22 +432,14 @@ func TestAsyncProducerRetryWithReferenceOpen(t *testing.T) {
518432
prodSuccess := new(ProduceResponse)
519433
prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
520434
leader.Returns(prodSuccess)
521-
select {
522-
case msg := <-producer.Errors():
523-
t.Error(msg.Err)
524-
case <-producer.Successes():
525-
}
435+
expectSuccesses(t, producer, 1)
526436

527437
// prime partition 1
528438
producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage)}
529439
prodSuccess = new(ProduceResponse)
530440
prodSuccess.AddTopicPartition("my_topic", 1, ErrNoError)
531441
leader.Returns(prodSuccess)
532-
select {
533-
case msg := <-producer.Errors():
534-
t.Error(msg.Err)
535-
case <-producer.Successes():
536-
}
442+
expectSuccesses(t, producer, 1)
537443

538444
// reboot the broker (the producer will get EOF on its existing connection)
539445
leader.Close()
@@ -549,11 +455,69 @@ func TestAsyncProducerRetryWithReferenceOpen(t *testing.T) {
549455
prodSuccess = new(ProduceResponse)
550456
prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
551457
leader.Returns(prodSuccess)
552-
select {
553-
case msg := <-producer.Errors():
554-
t.Error(msg.Err)
555-
case <-producer.Successes():
458+
expectSuccesses(t, producer, 1)
459+
460+
// shutdown
461+
closeProducer(t, producer)
462+
seedBroker.Close()
463+
leader.Close()
464+
}
465+
466+
func TestAsyncProducerFlusherRetryCondition(t *testing.T) {
467+
seedBroker := newMockBroker(t, 1)
468+
leader := newMockBroker(t, 2)
469+
470+
metadataResponse := new(MetadataResponse)
471+
metadataResponse.AddBroker(leader.Addr(), leader.BrokerID())
472+
metadataResponse.AddTopicPartition("my_topic", 0, leader.BrokerID(), nil, nil, ErrNoError)
473+
metadataResponse.AddTopicPartition("my_topic", 1, leader.BrokerID(), nil, nil, ErrNoError)
474+
seedBroker.Returns(metadataResponse)
475+
476+
config := NewConfig()
477+
config.Producer.Flush.Messages = 5
478+
config.Producer.Return.Successes = true
479+
config.Producer.Retry.Backoff = 0
480+
config.Producer.Retry.Max = 1
481+
config.Producer.Partitioner = NewManualPartitioner
482+
producer, err := NewAsyncProducer([]string{seedBroker.Addr()}, config)
483+
if err != nil {
484+
t.Fatal(err)
485+
}
486+
487+
// prime partitions
488+
for p := int32(0); p < 2; p++ {
489+
for i := 0; i < 5; i++ {
490+
producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage), Partition: p}
491+
}
492+
prodSuccess := new(ProduceResponse)
493+
prodSuccess.AddTopicPartition("my_topic", p, ErrNoError)
494+
leader.Returns(prodSuccess)
495+
expectSuccesses(t, producer, 5)
496+
}
497+
498+
// send more messages on partition 0
499+
for i := 0; i < 5; i++ {
500+
producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage), Partition: 0}
501+
}
502+
prodNotLeader := new(ProduceResponse)
503+
prodNotLeader.AddTopicPartition("my_topic", 0, ErrNotLeaderForPartition)
504+
leader.Returns(prodNotLeader)
505+
506+
// tell partition 0 to go to that broker again
507+
seedBroker.Returns(metadataResponse)
508+
509+
// succeed this time
510+
prodSuccess := new(ProduceResponse)
511+
prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
512+
leader.Returns(prodSuccess)
513+
expectSuccesses(t, producer, 5)
514+
515+
// put five more through
516+
for i := 0; i < 5; i++ {
517+
producer.Input() <- &ProducerMessage{Topic: "my_topic", Key: nil, Value: StringEncoder(TestMessage), Partition: 0}
556518
}
519+
leader.Returns(prodSuccess)
520+
expectSuccesses(t, producer, 5)
557521

558522
// shutdown
559523
closeProducer(t, producer)

0 commit comments

Comments
 (0)