Function TestRebalanceTooManyConsumers

reader_test.go:1376–1413

Test that a reader won't continually rebalance when there are more consumers than partitions in a group. https://github.com/segmentio/kafka-go/issues/200

(t *testing.T)

// than partitions in a group.
// https://github.com/segmentio/kafka-go/issues/200
func TestRebalanceTooManyConsumers(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	conf := ReaderConfig{
		Brokers: []string{"localhost:9092"},
		GroupID: makeGroupID(),
		Topic:   makeTopic(),
		MaxWait: time.Second,
	}

	// Create the first reader and wait for it to become the leader.
	r1 := NewReader(conf)

	// Give the reader some time to setup before reading a message
	time.Sleep(1 * time.Second)
	prepareReader(t, ctx, r1, makeTestSequence(1)...)

	_, err := r1.ReadMessage(ctx)
	if err != nil {
		t.Fatalf("failed to read message: %v", err)
	}
	// Clear the stats from the first rebalance.
	r1.Stats()

	// Second reader should cause one rebalance for each r1 and r2.
	r2 := NewReader(conf)

	// Wait for rebalances.
	time.Sleep(5 * time.Second)

	// Before the fix, r2 would cause continuous rebalances,
	// as it tried to handshake() repeatedly.
	rebalances := r1.Stats().Rebalances + r2.Stats().Rebalances
	if rebalances > 2 {
		t.Errorf("unexpected rebalances to first reader, got %d", rebalances)
	}
}

func TestConsumerGroupWithMissingTopic(t *testing.T) {
	t.Skip("this test doesn't work when the cluster is configured to auto-create topics")