Function TestClientRefreshesMetadataConcurrently

client_test.go:1117–1175

(t *testing.T)

}

func TestClientRefreshesMetadataConcurrently(t *testing.T) {
	seedBroker := NewMockBroker(t, 1)

	seedBroker.setHandler(func(req *request) (res encoderWithHeader) {
		mr, ok := req.body.(*MetadataRequest)
		if !ok {
			return nil
		}
		resp := new(MetadataResponse)
		for _, topic := range mr.Topics {
			switch topic {
			case "topic1":
				resp.Topics = append(resp.Topics, &TopicMetadata{
					Version:    1,
					Name:       "topic1",
					Partitions: []*PartitionMetadata{},
				})
			case "topic2":
				resp.Topics = append(resp.Topics, &TopicMetadata{
					Version:    1,
					Name:       "topic2",
					Partitions: []*PartitionMetadata{},
				})
			case "topic3":
				resp.Topics = append(resp.Topics, &TopicMetadata{
					Version: 1,
					Name:    "topic3",
					Err:     ErrUnknownTopicOrPartition,
				})
			default:
				t.Errorf("unexpected topic: %s", topic)
			}
		}
		resp.AddBroker(seedBroker.Addr(), 1)
		return resp
	})

	config := NewTestConfig()
	config.Metadata.SingleFlight = true
	client, err := NewClient([]string{seedBroker.Addr()}, config)
	require.NoError(t, err)

	var waitGroup sync.WaitGroup
	waitGroup.Add(100)
	for range 100 {
		go func() {
			defer waitGroup.Done()
			assert.NoError(t, client.RefreshMetadata("topic1"))
			assert.NoError(t, client.RefreshMetadata("topic2"))
			assert.ErrorIs(t, client.RefreshMetadata("topic3"), ErrUnknownTopicOrPartition)
			topics, err := client.Topics()
			assert.NoError(t, err)
			assert.Len(t, topics, 2)
		}()
	}
	waitGroup.Wait()
	safeClose(t, client)
	seedBroker.Close()