Function TestPartitionConsumerBrokerRace

consumer_test.go:1648–1722

(t *testing.T)

}

func TestPartitionConsumerBrokerRace(t *testing.T) {
	oldMaxProcs := runtime.GOMAXPROCS(2)
	defer runtime.GOMAXPROCS(oldMaxProcs)

	const iterations = 2048

	config := NewTestConfig()
	config.ChannelBufferSize = 0
	config.Consumer.MaxProcessingTime = time.Hour

	broker := &brokerConsumer{
		input: make(chan *brokerSubscription, 1),
		stop:  make(chan none),
	}

	child := &partitionConsumer{
		conf:           config,
		broker:         broker,
		messages:       make(chan *ConsumerMessage, 1),
		errors:         make(chan *ConsumerError, 1),
		feeder:         make(chan *partitionConsumerResponse, 1),
		trigger:        make(chan none, 1),
		dying:          make(chan none),
		dispatcherStop: make(chan none),
		topic:          "my_topic",
		partition:      0,
		fetchSize:      config.Consumer.Fetch.Default,
	}
	child.brokerSubscription = newBrokerSubscription(child)

	response := &FetchResponse{}
	response.AddMessage("my_topic", 0, nil, testMsg, 0)

	done := make(chan none)
	feederDone := make(chan none)
	messagesDone := make(chan none)

	go func() {
		defer close(feederDone)
		child.responseFeeder()
	}()

	go func() {
		defer close(messagesDone)
		for range child.messages {
		}
	}()

	go func() {
		for {
			select {
			case <-done:
				return
			default:
				child.broker = broker
				runtime.Gosched()
			}
		}