Function TestOffsetManagerFetchInitialLoadInProgress

offset_manager_test.go:521–561

Test fetchInitialOffset retry on ErrOffsetsLoadInProgress

(t *testing.T)

// Test fetchInitialOffset retry on ErrOffsetsLoadInProgress
func TestOffsetManagerFetchInitialLoadInProgress(t *testing.T) {
	var retryCount atomic.Int32
	backoff := func(retries, maxRetries int) time.Duration {
		retryCount.Add(1)
		return 0
	}
	om, testClient, broker, coordinator := initOffsetManagerWithBackoffFunc(t, 0, backoff, NewTestConfig())
	defer broker.Close()
	defer coordinator.Close()

	// Error on first fetchInitialOffset call
	responseBlock := OffsetFetchResponseBlock{
		Err:      ErrOffsetsLoadInProgress,
		Offset:   5,
		Metadata: "test_meta",
	}

	fetchResponse := new(OffsetFetchResponse)
	fetchResponse.AddBlock("my_topic", 0, &responseBlock)
	coordinator.Returns(fetchResponse)

	// Second fetchInitialOffset call is fine
	fetchResponse2 := new(OffsetFetchResponse)
	responseBlock2 := responseBlock
	responseBlock2.Err = ErrNoError
	fetchResponse2.AddBlock("my_topic", 0, &responseBlock2)
	coordinator.Returns(fetchResponse2)

	pom, err := om.ManagePartition("my_topic", 0)
	if err != nil {
		t.Error(err)
	}

	safeClose(t, pom)
	safeClose(t, om)
	safeClose(t, testClient)

	if retryCount.Load() == 0 {
		t.Fatal("Expected at least one retry")
	}
}

// fetchInitialOffset must retry when OffsetFetchResponse v2+ surfaces a
// retriable coordinator error at the top level with no per-partition blocks