client_test.go 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660
  1. package sarama
  2. import (
  3. "io"
  4. "sync"
  5. "testing"
  6. "time"
  7. )
  8. func safeClose(t testing.TB, c io.Closer) {
  9. err := c.Close()
  10. if err != nil {
  11. t.Error(err)
  12. }
  13. }
  14. func TestSimpleClient(t *testing.T) {
  15. seedBroker := NewMockBroker(t, 1)
  16. seedBroker.Returns(new(MetadataResponse))
  17. client, err := NewClient([]string{seedBroker.Addr()}, nil)
  18. if err != nil {
  19. t.Fatal(err)
  20. }
  21. seedBroker.Close()
  22. safeClose(t, client)
  23. }
  24. func TestCachedPartitions(t *testing.T) {
  25. seedBroker := NewMockBroker(t, 1)
  26. replicas := []int32{3, 1, 5}
  27. isr := []int32{5, 1}
  28. metadataResponse := new(MetadataResponse)
  29. metadataResponse.AddBroker("localhost:12345", 2)
  30. metadataResponse.AddTopicPartition("my_topic", 0, 2, replicas, isr, ErrNoError)
  31. metadataResponse.AddTopicPartition("my_topic", 1, 2, replicas, isr, ErrLeaderNotAvailable)
  32. seedBroker.Returns(metadataResponse)
  33. config := NewConfig()
  34. config.Metadata.Retry.Max = 0
  35. c, err := NewClient([]string{seedBroker.Addr()}, config)
  36. if err != nil {
  37. t.Fatal(err)
  38. }
  39. client := c.(*client)
  40. // Verify they aren't cached the same
  41. allP := client.cachedPartitionsResults["my_topic"][allPartitions]
  42. writeP := client.cachedPartitionsResults["my_topic"][writablePartitions]
  43. if len(allP) == len(writeP) {
  44. t.Fatal("Invalid lengths!")
  45. }
  46. tmp := client.cachedPartitionsResults["my_topic"]
  47. // Verify we actually use the cache at all!
  48. tmp[allPartitions] = []int32{1, 2, 3, 4}
  49. client.cachedPartitionsResults["my_topic"] = tmp
  50. if 4 != len(client.cachedPartitions("my_topic", allPartitions)) {
  51. t.Fatal("Not using the cache!")
  52. }
  53. seedBroker.Close()
  54. safeClose(t, client)
  55. }
  56. func TestClientDoesntCachePartitionsForTopicsWithErrors(t *testing.T) {
  57. seedBroker := NewMockBroker(t, 1)
  58. replicas := []int32{seedBroker.BrokerID()}
  59. metadataResponse := new(MetadataResponse)
  60. metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
  61. metadataResponse.AddTopicPartition("my_topic", 1, replicas[0], replicas, replicas, ErrNoError)
  62. metadataResponse.AddTopicPartition("my_topic", 2, replicas[0], replicas, replicas, ErrNoError)
  63. seedBroker.Returns(metadataResponse)
  64. config := NewConfig()
  65. config.Metadata.Retry.Max = 0
  66. client, err := NewClient([]string{seedBroker.Addr()}, config)
  67. if err != nil {
  68. t.Fatal(err)
  69. }
  70. metadataResponse = new(MetadataResponse)
  71. metadataResponse.AddTopic("unknown", ErrUnknownTopicOrPartition)
  72. seedBroker.Returns(metadataResponse)
  73. partitions, err := client.Partitions("unknown")
  74. if err != ErrUnknownTopicOrPartition {
  75. t.Error("Expected ErrUnknownTopicOrPartition, found", err)
  76. }
  77. if partitions != nil {
  78. t.Errorf("Should return nil as partition list, found %v", partitions)
  79. }
  80. // Should still use the cache of a known topic
  81. partitions, err = client.Partitions("my_topic")
  82. if err != nil {
  83. t.Errorf("Expected no error, found %v", err)
  84. }
  85. metadataResponse = new(MetadataResponse)
  86. metadataResponse.AddTopic("unknown", ErrUnknownTopicOrPartition)
  87. seedBroker.Returns(metadataResponse)
  88. // Should not use cache for unknown topic
  89. partitions, err = client.Partitions("unknown")
  90. if err != ErrUnknownTopicOrPartition {
  91. t.Error("Expected ErrUnknownTopicOrPartition, found", err)
  92. }
  93. if partitions != nil {
  94. t.Errorf("Should return nil as partition list, found %v", partitions)
  95. }
  96. seedBroker.Close()
  97. safeClose(t, client)
  98. }
  99. func TestClientSeedBrokers(t *testing.T) {
  100. seedBroker := NewMockBroker(t, 1)
  101. metadataResponse := new(MetadataResponse)
  102. metadataResponse.AddBroker("localhost:12345", 2)
  103. seedBroker.Returns(metadataResponse)
  104. client, err := NewClient([]string{seedBroker.Addr()}, nil)
  105. if err != nil {
  106. t.Fatal(err)
  107. }
  108. seedBroker.Close()
  109. safeClose(t, client)
  110. }
  111. func TestClientMetadata(t *testing.T) {
  112. seedBroker := NewMockBroker(t, 1)
  113. leader := NewMockBroker(t, 5)
  114. replicas := []int32{3, 1, 5}
  115. isr := []int32{5, 1}
  116. metadataResponse := new(MetadataResponse)
  117. metadataResponse.AddBroker(leader.Addr(), leader.BrokerID())
  118. metadataResponse.AddTopicPartition("my_topic", 0, leader.BrokerID(), replicas, isr, ErrNoError)
  119. metadataResponse.AddTopicPartition("my_topic", 1, leader.BrokerID(), replicas, isr, ErrLeaderNotAvailable)
  120. seedBroker.Returns(metadataResponse)
  121. config := NewConfig()
  122. config.Metadata.Retry.Max = 0
  123. client, err := NewClient([]string{seedBroker.Addr()}, config)
  124. if err != nil {
  125. t.Fatal(err)
  126. }
  127. topics, err := client.Topics()
  128. if err != nil {
  129. t.Error(err)
  130. } else if len(topics) != 1 || topics[0] != "my_topic" {
  131. t.Error("Client returned incorrect topics:", topics)
  132. }
  133. parts, err := client.Partitions("my_topic")
  134. if err != nil {
  135. t.Error(err)
  136. } else if len(parts) != 2 || parts[0] != 0 || parts[1] != 1 {
  137. t.Error("Client returned incorrect partitions for my_topic:", parts)
  138. }
  139. parts, err = client.WritablePartitions("my_topic")
  140. if err != nil {
  141. t.Error(err)
  142. } else if len(parts) != 1 || parts[0] != 0 {
  143. t.Error("Client returned incorrect writable partitions for my_topic:", parts)
  144. }
  145. tst, err := client.Leader("my_topic", 0)
  146. if err != nil {
  147. t.Error(err)
  148. } else if tst.ID() != 5 {
  149. t.Error("Leader for my_topic had incorrect ID.")
  150. }
  151. replicas, err = client.Replicas("my_topic", 0)
  152. if err != nil {
  153. t.Error(err)
  154. } else if replicas[0] != 3 {
  155. t.Error("Incorrect (or sorted) replica")
  156. } else if replicas[1] != 1 {
  157. t.Error("Incorrect (or sorted) replica")
  158. } else if replicas[2] != 5 {
  159. t.Error("Incorrect (or sorted) replica")
  160. }
  161. isr, err = client.InSyncReplicas("my_topic", 0)
  162. if err != nil {
  163. t.Error(err)
  164. } else if len(isr) != 2 {
  165. t.Error("Client returned incorrect ISRs for partition:", isr)
  166. } else if isr[0] != 5 {
  167. t.Error("Incorrect (or sorted) ISR:", isr)
  168. } else if isr[1] != 1 {
  169. t.Error("Incorrect (or sorted) ISR:", isr)
  170. }
  171. leader.Close()
  172. seedBroker.Close()
  173. safeClose(t, client)
  174. }
  175. func TestClientGetOffset(t *testing.T) {
  176. seedBroker := NewMockBroker(t, 1)
  177. leader := NewMockBroker(t, 2)
  178. leaderAddr := leader.Addr()
  179. metadata := new(MetadataResponse)
  180. metadata.AddTopicPartition("foo", 0, leader.BrokerID(), nil, nil, ErrNoError)
  181. metadata.AddBroker(leaderAddr, leader.BrokerID())
  182. seedBroker.Returns(metadata)
  183. client, err := NewClient([]string{seedBroker.Addr()}, nil)
  184. if err != nil {
  185. t.Fatal(err)
  186. }
  187. offsetResponse := new(OffsetResponse)
  188. offsetResponse.AddTopicPartition("foo", 0, 123)
  189. leader.Returns(offsetResponse)
  190. offset, err := client.GetOffset("foo", 0, OffsetNewest)
  191. if err != nil {
  192. t.Error(err)
  193. }
  194. if offset != 123 {
  195. t.Error("Unexpected offset, got ", offset)
  196. }
  197. leader.Close()
  198. seedBroker.Returns(metadata)
  199. leader = NewMockBrokerAddr(t, 2, leaderAddr)
  200. offsetResponse = new(OffsetResponse)
  201. offsetResponse.AddTopicPartition("foo", 0, 456)
  202. leader.Returns(offsetResponse)
  203. offset, err = client.GetOffset("foo", 0, OffsetNewest)
  204. if err != nil {
  205. t.Error(err)
  206. }
  207. if offset != 456 {
  208. t.Error("Unexpected offset, got ", offset)
  209. }
  210. seedBroker.Close()
  211. leader.Close()
  212. safeClose(t, client)
  213. }
  214. func TestClientReceivingUnknownTopic(t *testing.T) {
  215. seedBroker := NewMockBroker(t, 1)
  216. metadataResponse1 := new(MetadataResponse)
  217. seedBroker.Returns(metadataResponse1)
  218. config := NewConfig()
  219. config.Metadata.Retry.Max = 1
  220. config.Metadata.Retry.Backoff = 0
  221. client, err := NewClient([]string{seedBroker.Addr()}, config)
  222. if err != nil {
  223. t.Fatal(err)
  224. }
  225. metadataUnknownTopic := new(MetadataResponse)
  226. metadataUnknownTopic.AddTopic("new_topic", ErrUnknownTopicOrPartition)
  227. seedBroker.Returns(metadataUnknownTopic)
  228. seedBroker.Returns(metadataUnknownTopic)
  229. if err := client.RefreshMetadata("new_topic"); err != ErrUnknownTopicOrPartition {
  230. t.Error("ErrUnknownTopicOrPartition expected, got", err)
  231. }
  232. // If we are asking for the leader of a partition of the non-existing topic.
  233. // we will request metadata again.
  234. seedBroker.Returns(metadataUnknownTopic)
  235. seedBroker.Returns(metadataUnknownTopic)
  236. if _, err = client.Leader("new_topic", 1); err != ErrUnknownTopicOrPartition {
  237. t.Error("Expected ErrUnknownTopicOrPartition, got", err)
  238. }
  239. safeClose(t, client)
  240. seedBroker.Close()
  241. }
  242. func TestClientReceivingPartialMetadata(t *testing.T) {
  243. seedBroker := NewMockBroker(t, 1)
  244. leader := NewMockBroker(t, 5)
  245. metadataResponse1 := new(MetadataResponse)
  246. metadataResponse1.AddBroker(leader.Addr(), leader.BrokerID())
  247. seedBroker.Returns(metadataResponse1)
  248. config := NewConfig()
  249. config.Metadata.Retry.Max = 0
  250. client, err := NewClient([]string{seedBroker.Addr()}, config)
  251. if err != nil {
  252. t.Fatal(err)
  253. }
  254. replicas := []int32{leader.BrokerID(), seedBroker.BrokerID()}
  255. metadataPartial := new(MetadataResponse)
  256. metadataPartial.AddTopic("new_topic", ErrLeaderNotAvailable)
  257. metadataPartial.AddTopicPartition("new_topic", 0, leader.BrokerID(), replicas, replicas, ErrNoError)
  258. metadataPartial.AddTopicPartition("new_topic", 1, -1, replicas, []int32{}, ErrLeaderNotAvailable)
  259. seedBroker.Returns(metadataPartial)
  260. if err := client.RefreshMetadata("new_topic"); err != nil {
  261. t.Error("ErrLeaderNotAvailable should not make RefreshMetadata respond with an error")
  262. }
  263. // Even though the metadata was incomplete, we should be able to get the leader of a partition
  264. // for which we did get a useful response, without doing additional requests.
  265. partition0Leader, err := client.Leader("new_topic", 0)
  266. if err != nil {
  267. t.Error(err)
  268. } else if partition0Leader.Addr() != leader.Addr() {
  269. t.Error("Unexpected leader returned", partition0Leader.Addr())
  270. }
  271. // If we are asking for the leader of a partition that didn't have a leader before,
  272. // we will do another metadata request.
  273. seedBroker.Returns(metadataPartial)
  274. // Still no leader for the partition, so asking for it should return an error.
  275. _, err = client.Leader("new_topic", 1)
  276. if err != ErrLeaderNotAvailable {
  277. t.Error("Expected ErrLeaderNotAvailable, got", err)
  278. }
  279. safeClose(t, client)
  280. seedBroker.Close()
  281. leader.Close()
  282. }
  283. func TestClientRefreshBehaviour(t *testing.T) {
  284. seedBroker := NewMockBroker(t, 1)
  285. leader := NewMockBroker(t, 5)
  286. metadataResponse1 := new(MetadataResponse)
  287. metadataResponse1.AddBroker(leader.Addr(), leader.BrokerID())
  288. seedBroker.Returns(metadataResponse1)
  289. metadataResponse2 := new(MetadataResponse)
  290. metadataResponse2.AddTopicPartition("my_topic", 0xb, leader.BrokerID(), nil, nil, ErrNoError)
  291. seedBroker.Returns(metadataResponse2)
  292. client, err := NewClient([]string{seedBroker.Addr()}, nil)
  293. if err != nil {
  294. t.Fatal(err)
  295. }
  296. parts, err := client.Partitions("my_topic")
  297. if err != nil {
  298. t.Error(err)
  299. } else if len(parts) != 1 || parts[0] != 0xb {
  300. t.Error("Client returned incorrect partitions for my_topic:", parts)
  301. }
  302. tst, err := client.Leader("my_topic", 0xb)
  303. if err != nil {
  304. t.Error(err)
  305. } else if tst.ID() != 5 {
  306. t.Error("Leader for my_topic had incorrect ID.")
  307. }
  308. leader.Close()
  309. seedBroker.Close()
  310. safeClose(t, client)
  311. }
  312. func TestClientResurrectDeadSeeds(t *testing.T) {
  313. initialSeed := NewMockBroker(t, 0)
  314. emptyMetadata := new(MetadataResponse)
  315. initialSeed.Returns(emptyMetadata)
  316. conf := NewConfig()
  317. conf.Metadata.Retry.Backoff = 0
  318. conf.Metadata.RefreshFrequency = 0
  319. c, err := NewClient([]string{initialSeed.Addr()}, conf)
  320. if err != nil {
  321. t.Fatal(err)
  322. }
  323. initialSeed.Close()
  324. client := c.(*client)
  325. seed1 := NewMockBroker(t, 1)
  326. seed2 := NewMockBroker(t, 2)
  327. seed3 := NewMockBroker(t, 3)
  328. addr1 := seed1.Addr()
  329. addr2 := seed2.Addr()
  330. addr3 := seed3.Addr()
  331. // Overwrite the seed brokers with a fixed ordering to make this test deterministic.
  332. safeClose(t, client.seedBrokers[0])
  333. client.seedBrokers = []*Broker{NewBroker(addr1), NewBroker(addr2), NewBroker(addr3)}
  334. client.deadSeeds = []*Broker{}
  335. wg := sync.WaitGroup{}
  336. wg.Add(1)
  337. go func() {
  338. if err := client.RefreshMetadata(); err != nil {
  339. t.Error(err)
  340. }
  341. wg.Done()
  342. }()
  343. seed1.Close()
  344. seed2.Close()
  345. seed1 = NewMockBrokerAddr(t, 1, addr1)
  346. seed2 = NewMockBrokerAddr(t, 2, addr2)
  347. seed3.Close()
  348. seed1.Close()
  349. seed2.Returns(emptyMetadata)
  350. wg.Wait()
  351. if len(client.seedBrokers) != 2 {
  352. t.Error("incorrect number of live seeds")
  353. }
  354. if len(client.deadSeeds) != 1 {
  355. t.Error("incorrect number of dead seeds")
  356. }
  357. safeClose(t, c)
  358. }
  359. func TestClientController(t *testing.T) {
  360. seedBroker := NewMockBroker(t, 1)
  361. defer seedBroker.Close()
  362. controllerBroker := NewMockBroker(t, 2)
  363. defer controllerBroker.Close()
  364. seedBroker.SetHandlerByMap(map[string]MockResponse{
  365. "MetadataRequest": NewMockMetadataResponse(t).
  366. SetController(controllerBroker.BrokerID()).
  367. SetBroker(seedBroker.Addr(), seedBroker.BrokerID()).
  368. SetBroker(controllerBroker.Addr(), controllerBroker.BrokerID()),
  369. })
  370. cfg := NewConfig()
  371. // test kafka version greater than 0.10.0.0
  372. cfg.Version = V0_10_0_0
  373. client1, err := NewClient([]string{seedBroker.Addr()}, cfg)
  374. if err != nil {
  375. t.Fatal(err)
  376. }
  377. defer safeClose(t, client1)
  378. broker, err := client1.Controller()
  379. if err != nil {
  380. t.Fatal(err)
  381. }
  382. if broker.Addr() != controllerBroker.Addr() {
  383. t.Errorf("Expected controller to have address %s, found %s", controllerBroker.Addr(), broker.Addr())
  384. }
  385. // test kafka version earlier than 0.10.0.0
  386. cfg.Version = V0_9_0_1
  387. client2, err := NewClient([]string{seedBroker.Addr()}, cfg)
  388. if err != nil {
  389. t.Fatal(err)
  390. }
  391. defer safeClose(t, client2)
  392. if _, err = client2.Controller(); err != ErrControllerNotAvailable {
  393. t.Errorf("Expected Contoller() to return %s, found %s", ErrControllerNotAvailable, err)
  394. }
  395. }
  396. func TestClientCoordinatorWithConsumerOffsetsTopic(t *testing.T) {
  397. seedBroker := NewMockBroker(t, 1)
  398. staleCoordinator := NewMockBroker(t, 2)
  399. freshCoordinator := NewMockBroker(t, 3)
  400. replicas := []int32{staleCoordinator.BrokerID(), freshCoordinator.BrokerID()}
  401. metadataResponse1 := new(MetadataResponse)
  402. metadataResponse1.AddBroker(staleCoordinator.Addr(), staleCoordinator.BrokerID())
  403. metadataResponse1.AddBroker(freshCoordinator.Addr(), freshCoordinator.BrokerID())
  404. metadataResponse1.AddTopicPartition("__consumer_offsets", 0, replicas[0], replicas, replicas, ErrNoError)
  405. seedBroker.Returns(metadataResponse1)
  406. client, err := NewClient([]string{seedBroker.Addr()}, nil)
  407. if err != nil {
  408. t.Fatal(err)
  409. }
  410. coordinatorResponse1 := new(ConsumerMetadataResponse)
  411. coordinatorResponse1.Err = ErrConsumerCoordinatorNotAvailable
  412. seedBroker.Returns(coordinatorResponse1)
  413. coordinatorResponse2 := new(ConsumerMetadataResponse)
  414. coordinatorResponse2.CoordinatorID = staleCoordinator.BrokerID()
  415. coordinatorResponse2.CoordinatorHost = "127.0.0.1"
  416. coordinatorResponse2.CoordinatorPort = staleCoordinator.Port()
  417. seedBroker.Returns(coordinatorResponse2)
  418. broker, err := client.Coordinator("my_group")
  419. if err != nil {
  420. t.Error(err)
  421. }
  422. if staleCoordinator.Addr() != broker.Addr() {
  423. t.Errorf("Expected coordinator to have address %s, found %s", staleCoordinator.Addr(), broker.Addr())
  424. }
  425. if staleCoordinator.BrokerID() != broker.ID() {
  426. t.Errorf("Expected coordinator to have ID %d, found %d", staleCoordinator.BrokerID(), broker.ID())
  427. }
  428. // Grab the cached value
  429. broker2, err := client.Coordinator("my_group")
  430. if err != nil {
  431. t.Error(err)
  432. }
  433. if broker2.Addr() != broker.Addr() {
  434. t.Errorf("Expected the coordinator to be the same, but found %s vs. %s", broker2.Addr(), broker.Addr())
  435. }
  436. coordinatorResponse3 := new(ConsumerMetadataResponse)
  437. coordinatorResponse3.CoordinatorID = freshCoordinator.BrokerID()
  438. coordinatorResponse3.CoordinatorHost = "127.0.0.1"
  439. coordinatorResponse3.CoordinatorPort = freshCoordinator.Port()
  440. seedBroker.Returns(coordinatorResponse3)
  441. // Refresh the locally cahced value because it's stale
  442. if err := client.RefreshCoordinator("my_group"); err != nil {
  443. t.Error(err)
  444. }
  445. // Grab the fresh value
  446. broker3, err := client.Coordinator("my_group")
  447. if err != nil {
  448. t.Error(err)
  449. }
  450. if broker3.Addr() != freshCoordinator.Addr() {
  451. t.Errorf("Expected the freshCoordinator to be returned, but found %s.", broker3.Addr())
  452. }
  453. freshCoordinator.Close()
  454. staleCoordinator.Close()
  455. seedBroker.Close()
  456. safeClose(t, client)
  457. }
  458. func TestClientCoordinatorWithoutConsumerOffsetsTopic(t *testing.T) {
  459. seedBroker := NewMockBroker(t, 1)
  460. coordinator := NewMockBroker(t, 2)
  461. metadataResponse1 := new(MetadataResponse)
  462. seedBroker.Returns(metadataResponse1)
  463. config := NewConfig()
  464. config.Metadata.Retry.Max = 1
  465. config.Metadata.Retry.Backoff = 0
  466. client, err := NewClient([]string{seedBroker.Addr()}, config)
  467. if err != nil {
  468. t.Fatal(err)
  469. }
  470. coordinatorResponse1 := new(ConsumerMetadataResponse)
  471. coordinatorResponse1.Err = ErrConsumerCoordinatorNotAvailable
  472. seedBroker.Returns(coordinatorResponse1)
  473. metadataResponse2 := new(MetadataResponse)
  474. metadataResponse2.AddTopic("__consumer_offsets", ErrUnknownTopicOrPartition)
  475. seedBroker.Returns(metadataResponse2)
  476. replicas := []int32{coordinator.BrokerID()}
  477. metadataResponse3 := new(MetadataResponse)
  478. metadataResponse3.AddTopicPartition("__consumer_offsets", 0, replicas[0], replicas, replicas, ErrNoError)
  479. seedBroker.Returns(metadataResponse3)
  480. coordinatorResponse2 := new(ConsumerMetadataResponse)
  481. coordinatorResponse2.CoordinatorID = coordinator.BrokerID()
  482. coordinatorResponse2.CoordinatorHost = "127.0.0.1"
  483. coordinatorResponse2.CoordinatorPort = coordinator.Port()
  484. seedBroker.Returns(coordinatorResponse2)
  485. broker, err := client.Coordinator("my_group")
  486. if err != nil {
  487. t.Error(err)
  488. }
  489. if coordinator.Addr() != broker.Addr() {
  490. t.Errorf("Expected coordinator to have address %s, found %s", coordinator.Addr(), broker.Addr())
  491. }
  492. if coordinator.BrokerID() != broker.ID() {
  493. t.Errorf("Expected coordinator to have ID %d, found %d", coordinator.BrokerID(), broker.ID())
  494. }
  495. coordinator.Close()
  496. seedBroker.Close()
  497. safeClose(t, client)
  498. }
  499. func TestClientAutorefreshShutdownRace(t *testing.T) {
  500. seedBroker := NewMockBroker(t, 1)
  501. metadataResponse := new(MetadataResponse)
  502. seedBroker.Returns(metadataResponse)
  503. conf := NewConfig()
  504. conf.Metadata.RefreshFrequency = 100 * time.Millisecond
  505. client, err := NewClient([]string{seedBroker.Addr()}, conf)
  506. if err != nil {
  507. t.Fatal(err)
  508. }
  509. // Wait for the background refresh to kick in
  510. time.Sleep(110 * time.Millisecond)
  511. done := make(chan none)
  512. go func() {
  513. // Close the client
  514. if err := client.Close(); err != nil {
  515. t.Fatal(err)
  516. }
  517. close(done)
  518. }()
  519. // Wait for the Close to kick in
  520. time.Sleep(10 * time.Millisecond)
  521. // Then return some metadata to the still-running background thread
  522. leader := NewMockBroker(t, 2)
  523. metadataResponse.AddBroker(leader.Addr(), leader.BrokerID())
  524. metadataResponse.AddTopicPartition("foo", 0, leader.BrokerID(), []int32{2}, []int32{2}, ErrNoError)
  525. seedBroker.Returns(metadataResponse)
  526. <-done
  527. seedBroker.Close()
  528. // give the update time to happen so we get a panic if it's still running (which it shouldn't)
  529. time.Sleep(10 * time.Millisecond)
  530. }