Function TestOffsetManagerCommitConcurrentDeadlock

offset_manager_test.go:216–343

Test that concurrent Commit() calls do not deadlock when the coordinator returns errors that trigger releaseCoordinator while other goroutines are in constructRequest or coordinator(). This reproduces the lock ordering issue from https://github.com/IBM/sarama/issues/3191 where four goroutines form a

(t *testing.T)

// issue from https://github.com/IBM/sarama/issues/3191 where four goroutines
// form a cycle: broker.lock -> pomsLock -> brokerLock -> broker.lock.
func TestOffsetManagerCommitConcurrentDeadlock(t *testing.T) {
	seedBroker := NewMockBroker(t, 1)
	defer seedBroker.Close()

	coordinatorBroker := NewMockBroker(t, 2)
	defer coordinatorBroker.Close()

	findCoordDelay := 500 * time.Microsecond
	seedBroker.SetHandlerFuncByMap(map[string]requestHandlerFunc{
		"MetadataRequest": func(req *request) encoderWithHeader {
			resp := new(MetadataResponse)
			resp.AddBroker(coordinatorBroker.Addr(), coordinatorBroker.BrokerID())
			return resp
		},
		"FindCoordinatorRequest": func(req *request) encoderWithHeader {
			// slow coordinator lookup increases the window where brokerLock
			// is held in coordinator(), while LeastLoadedBroker calls
			// ResponseSize (which needs broker.lock), widening the
			// deadlock window
			time.Sleep(findCoordDelay)
			resp := new(FindCoordinatorResponse)
			resp.Coordinator = &Broker{id: coordinatorBroker.brokerID, addr: coordinatorBroker.Addr()}
			return resp
		},
	})

	var commitCount atomic.Int64
	coordinatorBroker.SetHandlerFuncByMap(map[string]requestHandlerFunc{
		"ConsumerMetadataRequest": func(req *request) encoderWithHeader {
			return &ConsumerMetadataResponse{
				CoordinatorID:   coordinatorBroker.BrokerID(),
				CoordinatorHost: "127.0.0.1",
				CoordinatorPort: coordinatorBroker.Port(),
			}
		},
		"FindCoordinatorRequest": func(req *request) encoderWithHeader {
			time.Sleep(findCoordDelay)
			resp := new(FindCoordinatorResponse)
			resp.Coordinator = &Broker{id: coordinatorBroker.brokerID, addr: coordinatorBroker.Addr()}
			return resp
		},
		"OffsetFetchRequest": func(r *request) encoderWithHeader {
			req := r.body.(*OffsetFetchRequest)
			resp := new(OffsetFetchResponse)
			resp.Blocks = map[string]map[int32]*OffsetFetchResponseBlock{}
			for topic, partitions := range req.partitions {
				for _, partition := range partitions {
					if _, ok := resp.Blocks[topic]; !ok {
						resp.Blocks[topic] = map[int32]*OffsetFetchResponseBlock{}
					}
					resp.Blocks[topic][partition] = &OffsetFetchResponseBlock{
						Offset: 0,
						Err:    ErrNoError,
					}
				}
			}
			return resp
		},