client_test.go 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151
  1. package sarama
  2. import (
  3. "io"
  4. "testing"
  5. )
  6. func safeClose(t *testing.T, c io.Closer) {
  7. err := c.Close()
  8. if err != nil {
  9. t.Error(err)
  10. }
  11. }
  12. func TestDefaultClientConfigValidates(t *testing.T) {
  13. config := NewClientConfig()
  14. if err := config.Validate(); err != nil {
  15. t.Error(err)
  16. }
  17. }
  18. func TestSimpleClient(t *testing.T) {
  19. mb := NewMockBroker(t, 1)
  20. mb.Returns(new(MetadataResponse))
  21. client, err := NewClient("client_id", []string{mb.Addr()}, nil)
  22. if err != nil {
  23. t.Fatal(err)
  24. }
  25. defer safeClose(t, client)
  26. defer mb.Close()
  27. }
  28. func TestClientSeedBrokers(t *testing.T) {
  29. mb1 := NewMockBroker(t, 1)
  30. mb2 := NewMockBroker(t, 2)
  31. mdr := new(MetadataResponse)
  32. mdr.AddBroker(mb2.Addr(), mb2.BrokerID())
  33. mb1.Returns(mdr)
  34. client, err := NewClient("client_id", []string{mb1.Addr()}, nil)
  35. if err != nil {
  36. t.Fatal(err)
  37. }
  38. defer safeClose(t, client)
  39. defer mb1.Close()
  40. defer mb2.Close()
  41. }
  42. func TestClientMetadata(t *testing.T) {
  43. mb1 := NewMockBroker(t, 1)
  44. mb5 := NewMockBroker(t, 5)
  45. replicas := []int32{3, 1, 5}
  46. isr := []int32{5, 1}
  47. mdr := new(MetadataResponse)
  48. mdr.AddBroker(mb5.Addr(), mb5.BrokerID())
  49. mdr.AddTopicPartition("my_topic", 0, mb5.BrokerID(), replicas, isr)
  50. mb1.Returns(mdr)
  51. client, err := NewClient("client_id", []string{mb1.Addr()}, nil)
  52. if err != nil {
  53. t.Fatal(err)
  54. }
  55. defer safeClose(t, client)
  56. defer mb1.Close()
  57. defer mb5.Close()
  58. topics, err := client.Topics()
  59. if err != nil {
  60. t.Error(err)
  61. } else if len(topics) != 1 || topics[0] != "my_topic" {
  62. t.Error("Client returned incorrect topics:", topics)
  63. }
  64. parts, err := client.Partitions("my_topic")
  65. if err != nil {
  66. t.Error(err)
  67. } else if len(parts) != 1 || parts[0] != 0 {
  68. t.Error("Client returned incorrect partitions for my_topic:", parts)
  69. }
  70. tst, err := client.Leader("my_topic", 0)
  71. if err != nil {
  72. t.Error(err)
  73. } else if tst.ID() != 5 {
  74. t.Error("Leader for my_topic had incorrect ID.")
  75. }
  76. replicas, err = client.Replicas("my_topic", 0)
  77. if err != nil {
  78. t.Error(err)
  79. } else if replicas[0] != 1 {
  80. t.Error("Incorrect (or unsorted) replica")
  81. } else if replicas[1] != 3 {
  82. t.Error("Incorrect (or unsorted) replica")
  83. } else if replicas[2] != 5 {
  84. t.Error("Incorrect (or unsorted) replica")
  85. }
  86. isr, err = client.ReplicasInSync("my_topic", 0)
  87. if err != nil {
  88. t.Error(err)
  89. } else if isr[0] != 1 {
  90. t.Error("Incorrect (or unsorted) isr")
  91. } else if isr[1] != 5 {
  92. t.Error("Incorrect (or unsorted) isr")
  93. }
  94. }
  95. func TestClientRefreshBehaviour(t *testing.T) {
  96. mb1 := NewMockBroker(t, 1)
  97. mb5 := NewMockBroker(t, 5)
  98. mdr := new(MetadataResponse)
  99. mdr.AddBroker(mb5.Addr(), mb5.BrokerID())
  100. mb1.Returns(mdr)
  101. mdr2 := new(MetadataResponse)
  102. mdr2.AddTopicPartition("my_topic", 0xb, mb5.BrokerID(), nil, nil)
  103. mb5.Returns(mdr2)
  104. client, err := NewClient("clientID", []string{mb1.Addr()}, nil)
  105. if err != nil {
  106. t.Fatal(err)
  107. }
  108. defer safeClose(t, client)
  109. defer mb1.Close()
  110. defer mb5.Close()
  111. parts, err := client.Partitions("my_topic")
  112. if err != nil {
  113. t.Error(err)
  114. } else if len(parts) != 1 || parts[0] != 0xb {
  115. t.Error("Client returned incorrect partitions for my_topic:", parts)
  116. }
  117. tst, err := client.Leader("my_topic", 0xb)
  118. if err != nil {
  119. t.Error(err)
  120. } else if tst.ID() != 5 {
  121. t.Error("Leader for my_topic had incorrect ID.")
  122. }
  123. client.disconnectBroker(tst)
  124. }