Function topicWithEvenLeaders

functional_admin_test.go:20–66

(t *testing.T, adminClient ClusterAdmin, client Client, numPartitions int32)

)

func topicWithEvenLeaders(t *testing.T, adminClient ClusterAdmin, client Client, numPartitions int32) (string, error) {
	t.Helper()

	if len(FunctionalTestEnv.KafkaBrokerAddrs) == 0 {
		return "", fmt.Errorf("no brokers available for replica assignment")
	}

	brokers := client.Brokers()
	brokerIDs := make([]int32, 0, len(brokers))
	for _, broker := range brokers {
		brokerIDs = append(brokerIDs, broker.ID())
	}
	if len(brokerIDs) == 0 {
		return "", fmt.Errorf("no broker IDs available for replica assignment")
	}
	slices.Sort(brokerIDs)

	topic := fmt.Sprintf("list-offsets-%d", time.Now().UnixNano())
	replicaAssignment := make(map[int32][]int32, numPartitions)
	for partition := int32(0); partition < numPartitions; partition++ {
		brokerIndex := partition % int32(len(brokerIDs))
		replicaAssignment[partition] = []int32{brokerIDs[brokerIndex]}
	}

	err := adminClient.CreateTopic(topic, &TopicDetail{
		NumPartitions:     -1,
		ReplicationFactor: -1,
		ReplicaAssignment: replicaAssignment,
	}, false)
	if err != nil {
		return "", err
	}

	// topic creation is asynchronous; wait for every partition to have an
	// elected leader before returning (producing too early fails with
	// ErrNotLeaderForPartition or ErrUnknownTopicOrPartition)
	require.EventuallyWithT(t, func(t *assert.CollectT) {
		require.NoError(t, client.RefreshMetadata(topic))
		for partition := int32(0); partition < numPartitions; partition++ {
			leader, err := client.Leader(topic, partition)
			require.NoError(t, err, "no leader for %s/%d", topic, partition)
			require.NotNil(t, leader)
		}
	}, 30*time.Second, 250*time.Millisecond, "leaders were not elected for all partitions of %s", topic)

	return topic, nil
}

func produceMessagesForPartitions(t *testing.T, client Client, topic string, partitionsCount int32, messagesPerPartition int, baseTimestamp int64) {
	t.Helper()