Function TestNewOffsetManagerOffsetsManualCommit

offset_manager_test.go:419–473

(t *testing.T)

}

func TestNewOffsetManagerOffsetsManualCommit(t *testing.T) {
	// Tests to validate configuration when `Consumer.Offsets.AutoCommit.Enable` is false
	config := NewTestConfig()
	config.Consumer.Offsets.AutoCommit.Enable = false

	om, testClient, broker, coordinator := initOffsetManagerWithBackoffFunc(t, 0, nil, config)
	defer broker.Close()
	defer coordinator.Close()
	pom := initPartitionOffsetManager(t, om, coordinator, 5, "original_meta")

	// Wait long enough for the test not to fail..
	timeout := 50 * config.Consumer.Offsets.AutoCommit.Interval

	ocResponse := new(OffsetCommitResponse)
	ocResponse.AddError("my_topic", 0, ErrNoError)
	called := make(chan none)
	handler := func(req *request) (res encoderWithHeader) {
		close(called)
		return ocResponse
	}
	coordinator.setHandler(handler)

	// Should not trigger an auto-commit
	expected := int64(1)
	pom.ResetOffset(expected, "modified_meta")
	_, _ = pom.NextOffset()

	select {
	case <-called:
		// OffsetManager called on the wire.
		t.Errorf("Received request when AutoCommit is disabled")
	case <-time.After(timeout):
		// Timeout waiting for OffsetManager to call on the wire.
		// OK
	}

	// Setup again to test manual commit
	called = make(chan none)

	om.Commit()

	select {
	case <-called:
		// OffsetManager called on the wire.
		// OK
	case <-time.After(timeout):
		// Timeout waiting for OffsetManager to call on the wire.
		t.Errorf("No request received for after waiting for %v", timeout)
	}

	// !! om must be closed before the pom so pom.release() is called before pom.Close()
	safeClose(t, om)
	safeClose(t, pom)
	safeClose(t, testClient)
}

// Test recovery from ErrNotCoordinatorForConsumer
// on first fetchInitialOffset call