Function TestClientUpdateMetadataErrorAndRetry

client_test.go:1060–1115

(t *testing.T)

}

func TestClientUpdateMetadataErrorAndRetry(t *testing.T) {
	seedBroker := NewMockBroker(t, 1)
	var called atomic.Int32

	seedBroker.setHandler(func(req *request) (res encoderWithHeader) {
		if req.body.key() != 3 {
			t.Error("this test sends only Metadata requests")
			return
		}
		resp := new(MetadataResponse)
		for _, topic := range req.body.(*MetadataRequest).Topics {
			if topic == "new_topic" {
				resp.Topics = append(resp.Topics, &TopicMetadata{
					Version: 1,
					Name:    "new_topic",
					Err:     ErrUnknownTopicOrPartition,
				})
			}
		}
		called.Add(1)
		resp.AddBroker(seedBroker.Addr(), 1)
		return resp
	})

	config := NewTestConfig()
	config.Metadata.Retry.Max = 3
	config.Metadata.Retry.Backoff = 200 * time.Millisecond
	config.Metadata.RefreshFrequency = 0
	config.Net.ReadTimeout = 10 * time.Millisecond
	config.Net.WriteTimeout = 10 * time.Millisecond
	config.Metadata.SingleFlight = true
	client, err := NewClient([]string{seedBroker.Addr()}, config)
	if err != nil {
		t.Fatal(err)
	}
	waitGroup := sync.WaitGroup{}
	waitGroup.Add(10)
	for range 10 {
		go func() {
			defer waitGroup.Done()
			err := client.RefreshMetadata("new_topic")
			if err == nil {
				t.Error("should return error")
				return
			}
		}()
	}
	waitGroup.Wait()
	safeClose(t, client)
	seedBroker.Close()
	// The refresh metadata is always for the same topic,
	// it should have been batched.
	if count := called.Load(); count >= 7 {
		t.Errorf("Refresh metadata was called %d times, this should be less than 7.", count)
	}
}

func TestClientRefreshesMetadataConcurrently(t *testing.T) {