sync_producer_test.go 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199
  1. package sarama
  2. import (
  3. "log"
  4. "sync"
  5. "testing"
  6. )
  7. func TestSyncProducer(t *testing.T) {
  8. seedBroker := NewMockBroker(t, 1)
  9. leader := NewMockBroker(t, 2)
  10. metadataResponse := new(MetadataResponse)
  11. metadataResponse.AddBroker(leader.Addr(), leader.BrokerID())
  12. metadataResponse.AddTopicPartition("my_topic", 0, leader.BrokerID(), nil, nil, ErrNoError)
  13. seedBroker.Returns(metadataResponse)
  14. prodSuccess := new(ProduceResponse)
  15. prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
  16. for i := 0; i < 10; i++ {
  17. leader.Returns(prodSuccess)
  18. }
  19. producer, err := NewSyncProducer([]string{seedBroker.Addr()}, nil)
  20. if err != nil {
  21. t.Fatal(err)
  22. }
  23. for i := 0; i < 10; i++ {
  24. msg := &ProducerMessage{
  25. Topic: "my_topic",
  26. Value: StringEncoder(TestMessage),
  27. Metadata: "test",
  28. }
  29. partition, offset, err := producer.SendMessage(msg)
  30. if partition != 0 || msg.Partition != partition {
  31. t.Error("Unexpected partition")
  32. }
  33. if offset != 0 || msg.Offset != offset {
  34. t.Error("Unexpected offset")
  35. }
  36. if str, ok := msg.Metadata.(string); !ok || str != "test" {
  37. t.Error("Unexpected metadata")
  38. }
  39. if err != nil {
  40. t.Error(err)
  41. }
  42. }
  43. safeClose(t, producer)
  44. leader.Close()
  45. seedBroker.Close()
  46. }
  47. func TestSyncProducerBatch(t *testing.T) {
  48. seedBroker := NewMockBroker(t, 1)
  49. leader := NewMockBroker(t, 2)
  50. metadataResponse := new(MetadataResponse)
  51. metadataResponse.AddBroker(leader.Addr(), leader.BrokerID())
  52. metadataResponse.AddTopicPartition("my_topic", 0, leader.BrokerID(), nil, nil, ErrNoError)
  53. seedBroker.Returns(metadataResponse)
  54. prodSuccess := new(ProduceResponse)
  55. prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
  56. leader.Returns(prodSuccess)
  57. config := NewConfig()
  58. config.Producer.Flush.Messages = 3
  59. config.Producer.Return.Successes = true
  60. producer, err := NewSyncProducer([]string{seedBroker.Addr()}, config)
  61. if err != nil {
  62. t.Fatal(err)
  63. }
  64. err = producer.SendMessages([]*ProducerMessage{
  65. {
  66. Topic: "my_topic",
  67. Value: StringEncoder(TestMessage),
  68. Metadata: "test",
  69. },
  70. {
  71. Topic: "my_topic",
  72. Value: StringEncoder(TestMessage),
  73. Metadata: "test",
  74. },
  75. {
  76. Topic: "my_topic",
  77. Value: StringEncoder(TestMessage),
  78. Metadata: "test",
  79. },
  80. })
  81. if err != nil {
  82. t.Error(err)
  83. }
  84. safeClose(t, producer)
  85. leader.Close()
  86. seedBroker.Close()
  87. }
  88. func TestConcurrentSyncProducer(t *testing.T) {
  89. seedBroker := NewMockBroker(t, 1)
  90. leader := NewMockBroker(t, 2)
  91. metadataResponse := new(MetadataResponse)
  92. metadataResponse.AddBroker(leader.Addr(), leader.BrokerID())
  93. metadataResponse.AddTopicPartition("my_topic", 0, leader.BrokerID(), nil, nil, ErrNoError)
  94. seedBroker.Returns(metadataResponse)
  95. prodSuccess := new(ProduceResponse)
  96. prodSuccess.AddTopicPartition("my_topic", 0, ErrNoError)
  97. leader.Returns(prodSuccess)
  98. config := NewConfig()
  99. config.Producer.Flush.Messages = 100
  100. config.Producer.Return.Successes = true
  101. producer, err := NewSyncProducer([]string{seedBroker.Addr()}, config)
  102. if err != nil {
  103. t.Fatal(err)
  104. }
  105. wg := sync.WaitGroup{}
  106. for i := 0; i < 100; i++ {
  107. wg.Add(1)
  108. go func() {
  109. msg := &ProducerMessage{Topic: "my_topic", Value: StringEncoder(TestMessage)}
  110. partition, _, err := producer.SendMessage(msg)
  111. if partition != 0 {
  112. t.Error("Unexpected partition")
  113. }
  114. if err != nil {
  115. t.Error(err)
  116. }
  117. wg.Done()
  118. }()
  119. }
  120. wg.Wait()
  121. safeClose(t, producer)
  122. leader.Close()
  123. seedBroker.Close()
  124. }
  125. func TestSyncProducerToNonExistingTopic(t *testing.T) {
  126. broker := NewMockBroker(t, 1)
  127. metadataResponse := new(MetadataResponse)
  128. metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
  129. metadataResponse.AddTopicPartition("my_topic", 0, broker.BrokerID(), nil, nil, ErrNoError)
  130. broker.Returns(metadataResponse)
  131. config := NewConfig()
  132. config.Metadata.Retry.Max = 0
  133. config.Producer.Retry.Max = 0
  134. config.Producer.Return.Successes = true
  135. producer, err := NewSyncProducer([]string{broker.Addr()}, config)
  136. if err != nil {
  137. t.Fatal(err)
  138. }
  139. metadataResponse = new(MetadataResponse)
  140. metadataResponse.AddTopic("unknown", ErrUnknownTopicOrPartition)
  141. broker.Returns(metadataResponse)
  142. _, _, err = producer.SendMessage(&ProducerMessage{Topic: "unknown"})
  143. if err != ErrUnknownTopicOrPartition {
  144. t.Error("Uxpected ErrUnknownTopicOrPartition, found:", err)
  145. }
  146. safeClose(t, producer)
  147. broker.Close()
  148. }
  149. // This example shows the basic usage pattern of the SyncProducer.
  150. func ExampleSyncProducer() {
  151. producer, err := NewSyncProducer([]string{"localhost:9092"}, nil)
  152. if err != nil {
  153. log.Fatalln(err)
  154. }
  155. defer func() {
  156. if err := producer.Close(); err != nil {
  157. log.Fatalln(err)
  158. }
  159. }()
  160. msg := &ProducerMessage{Topic: "my_topic", Value: StringEncoder("testing 123")}
  161. partition, offset, err := producer.SendMessage(msg)
  162. if err != nil {
  163. log.Printf("FAILED to send message: %s\n", err)
  164. } else {
  165. log.Printf("> message sent to partition %d at offset %d\n", partition, offset)
  166. }
  167. }