Function Test_handleThrottledResponse

broker_test.go:1608–1702

(t *testing.T)

}

func Test_handleThrottledResponse(t *testing.T) {
	mb := NewMockBroker(nil, 0)
	defer mb.Close()
	broker := NewBroker(mb.Addr())
	broker.id = 0
	conf := NewTestConfig()
	conf.Version = V1_0_0_0
	throttleTimeMs := 100
	throttleTime := time.Duration(throttleTimeMs) * time.Millisecond
	tests := []struct {
		name        string
		response    protocolBody
		expectDelay bool
	}{
		{
			name: "throttled response w/millisecond field",
			response: &MetadataResponse{
				ThrottleTimeMs: int32(throttleTimeMs),
			},
			expectDelay: true,
		},
		{
			name: "not throttled response w/millisecond field",
			response: &MetadataResponse{
				ThrottleTimeMs: 0,
			},
		},
		{
			name: "throttled response w/time.Duration field",
			response: &ProduceResponse{
				ThrottleTime: throttleTime,
			},
			expectDelay: true,
		},
		{
			name: "not throttled response w/time.Duration field",
			response: &ProduceResponse{
				ThrottleTime: time.Duration(0),
			},
		},
		{
			name:     "not throttled response with no throttle time field",
			response: &SaslHandshakeResponse{},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			broker.metricRegistry = metrics.NewRegistry()
			broker.brokerThrottleTime = broker.registerHistogram("throttle-time-in-ms")
			startTime := time.Now()
			broker.handleThrottledResponse(tt.response)
			broker.waitIfThrottled()
			if tt.expectDelay {
				if time.Since(startTime) < throttleTime {
					t.Fatal("expected throttling to cause delay")
				}
				if broker.brokerThrottleTime.Min() != int64(throttleTimeMs) {
					t.Fatal("expected throttling to update metrics")