Function TestPartitionConsumerDispatcherOrphanedClose

consumer_test.go:1727–1789

TestPartitionConsumerDispatcherOrphanedClose verifies the dispatcher exits after AsyncClose when the brokerConsumer has detached this child from its subscriptions and there is a pending redispatch on the trigger channel.

(t *testing.T)

// after AsyncClose when the brokerConsumer has detached this child from its
// subscriptions and there is a pending redispatch on the trigger channel.
func TestPartitionConsumerDispatcherOrphanedClose(t *testing.T) {
	newOrphan := func() *partitionConsumer {
		config := NewTestConfig()
		config.Consumer.Retry.Backoff = time.Hour
		config.Consumer.MaxWaitTime = 50 * time.Millisecond

		c := &consumer{
			conf:            config,
			children:        make(map[string]map[int32]*partitionConsumer),
			brokerConsumers: make(map[*Broker]*brokerConsumer),
		}
		bc := &brokerConsumer{consumer: c, input: make(chan *brokerSubscription), refs: 1, stop: make(chan none)}
		child := &partitionConsumer{
			consumer:       c,
			conf:           config,
			broker:         bc,
			feeder:         make(chan *partitionConsumerResponse, 1),
			trigger:        make(chan none, 1),
			dying:          make(chan none),
			dispatcherStop: make(chan none),
		}
		child.brokerSubscription = newBrokerSubscription(child)
		return child
	}

	runDispatcher := func(t *testing.T, child *partitionConsumer) {
		t.Helper()
		done := make(chan none)
		go func() {
			defer close(done)
			child.dispatcher()
		}()
		select {
		case <-done:
		case <-time.After(2 * time.Second):
			require.FailNow(t, "dispatcher hung after AsyncClose on orphaned child")
		}
	}

	t.Run("subscription released before dying", func(t *testing.T) {
		child := newOrphan()
		child.brokerSubscription.release()
		child.triggerRedispatch()
		close(child.dying)
		runDispatcher(t, child)
	})

	t.Run("subscription released after dying", func(t *testing.T) {
		child := newOrphan()
		child.triggerRedispatch()
		close(child.dying)
		child.brokerSubscription.release()
		runDispatcher(t, child)
	})

	t.Run("stopDispatcher unblocks waitForBrokerHandover", func(t *testing.T) {
		child := newOrphan()
		child.triggerRedispatch()