Function TestConsumerAbortNoGoroutineLeak

consumer_test.go:2377–2583

TestConsumerAbortNoGoroutineLeak verifies that brokerConsumer.abort() does not leak the subscriptionManager goroutine when children are already shutting down or already queued for redispatch.

(t *testing.T)

// not leak the subscriptionManager goroutine when children are already
// shutting down or already queued for redispatch.
func TestConsumerAbortNoGoroutineLeak(t *testing.T) {
	metrics.UseNilMetrics = true
	defer func() { metrics.UseNilMetrics = false }()

	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())

	config := NewTestConfig()
	config.Consumer.Return.Errors = true

	broker0 := NewMockBroker(t, 0)
	defer broker0.Close()

	broker0.SetHandlerByMap(map[string]MockResponse{
		"MetadataRequest": NewMockMetadataResponse(t).
			SetBroker(broker0.Addr(), broker0.BrokerID()).
			SetLeader("my_topic", 0, broker0.BrokerID()),
		"OffsetRequest": NewMockOffsetResponse(t).
			SetOffset("my_topic", 0, OffsetOldest, 0).
			SetOffset("my_topic", 0, OffsetNewest, 1000),
	})

	client, err := NewClient([]string{broker0.Addr()}, config)
	require.NoError(t, err)
	defer client.Close()

	realBroker := client.Brokers()[0]

	newChild := func(errorsBuffer int) *partitionConsumer {
		return &partitionConsumer{
			conf:           config,
			topic:          "my_topic",
			partition:      0,
			trigger:        make(chan none, 1),
			dying:          make(chan none),
			dispatcherStop: make(chan none),
			messages:       make(chan *ConsumerMessage, config.ChannelBufferSize),
			errors:         make(chan *ConsumerError, errorsBuffer),
			feeder:         make(chan *partitionConsumerResponse, 1),
		}
	}

	newBrokerConsumer := func(child *partitionConsumer) *brokerConsumer {
		c := &consumer{
			client:          client,
			conf:            config,
			children:        make(map[string]map[int32]*partitionConsumer),
			brokerConsumers: make(map[*Broker]*brokerConsumer),
			metricRegistry:  newCleanupRegistry(config.MetricRegistry),
		}
		child.consumer = c
		subscription := newBrokerSubscription(child)

		bc := &brokerConsumer{
			consumer:         c,
			broker:           realBroker,
			input:            make(chan *brokerSubscription),
			newSubscriptions: make(chan []*brokerSubscription),
			subscriptions:    map[*partitionConsumer]*brokerSubscription{child: subscription},