Function TestConsume_RaceTest

consumer_group_test.go:121–200

(t *testing.T)

}

func TestConsume_RaceTest(t *testing.T) {
	const (
		groupID     = "test-group"
		topic       = "test-topic"
		offsetStart = int64(1234)
	)

	cfg := NewTestConfig()
	cfg.Version = V2_8_1_0
	cfg.Consumer.Return.Errors = true
	cfg.Metadata.Full = true

	seedBroker := NewMockBroker(t, 1)
	defer seedBroker.Close()

	handlerMap := map[string]MockResponse{
		"ApiVersionsRequest": NewMockApiVersionsResponse(t),
		"MetadataRequest": NewMockMetadataResponse(t).
			SetBroker(seedBroker.Addr(), seedBroker.BrokerID()).
			SetError("mismatched-topic", ErrUnknownTopicOrPartition),
		"OffsetRequest": NewMockOffsetResponse(t).
			SetOffset(topic, 0, -1, offsetStart),
		"OffsetFetchRequest": NewMockOffsetFetchResponse(t).
			SetOffset(groupID, topic, 0, offsetStart, "", ErrNoError),
		"FindCoordinatorRequest": NewMockFindCoordinatorResponse(t).
			SetCoordinator(CoordinatorGroup, groupID, seedBroker),
		"JoinGroupRequest": NewMockJoinGroupResponse(t),
		"SyncGroupRequest": NewMockSyncGroupResponse(t).SetMemberAssignment(
			&ConsumerGroupMemberAssignment{
				Version:  1,
				Topics:   map[string][]int32{topic: {0}}, // map "test-topic" to partition 0
				UserData: []byte{0x01},
			},
		),
		"HeartbeatRequest": NewMockHeartbeatResponse(t),
	}
	seedBroker.SetHandlerByMap(handlerMap)

	cancelCtx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second))

	retryWait := 10 * time.Millisecond
	var err error
	clientRetries := 0
outerFor:
	for {
		_, err = NewConsumerGroup([]string{seedBroker.Addr()}, groupID, cfg)
		if err == nil {
			break
		}

		if retryWait < time.Minute {
			retryWait *= 2
		}

		clientRetries++

		timer := time.NewTimer(retryWait)
		select {