[KIP-460] Elect leader api implemented

[PratRanj07]

Implemented the Elect Leaders API(KIP - 460) and wrote unit tests for it.

The build is gonna fail until the ElectLeaders api is merged in ibrdkafka.

Did the manual testing for various cases:

Single Partition:

dotnet run localhost:9092 elect-leaders 0 quickstart-events 0

One or more elect leaders operations failed.
An error occurred in elect leaders operation in Topic quickstart-events Partition [0]: Code: ElectionNotNeeded, Reason: Leader election not needed for topic partition.

Multiple Partition:

dotnet run localhost:9092 elect-leaders 0 quickstart-events 0 first_topic 0

One or more elect leaders operations failed.
An error occurred in elect leaders operation in Topic quickstart-events Partition [0]: Code: ElectionNotNeeded, Reason: Leader election not needed for topic partition.
An error occurred in elect leaders operation in Topic first_topic Partition [0]: Code: ElectionNotNeeded, Reason: Leader election not needed for topic partition.

Empty Partitions is gonna give us an empty list.

Null partitions perform elections for all the partitions but the results only consists of success cases and cases where error is other than ElectionNotNeeded.

which is the desired result in case Leader Election is not required in the partitions.

Unit tests also ran fine.

`dotnet test --filter "FullyQualifiedName~Confluent.Kafka.UnitTests.ElectLeadersErrorTests"

Passed! - Failed: 0, Passed: 9, Skipped: 0, Total: 9, Duration: 3 s - Confluent.Kafka.UnitTests.dll (net6.0)

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[anchitj]

Rename to topicPartitionsPtr and topicPartitionsCountPtr

[anchitj]

topicPartitionsPtrArr

[anchitj]

Rename to overallError

[anchitj]

Do we need to add a check here for if overall err is NoError first? Otherwise we'd always overwrite it

[anchitj]

Also add null checks in extraction topicPartitionErrorsPtr non null, topicPartitionErrorsCount > 0 and in the loop topicPartitionErrorsPtr[i] not IntPtr.Zero

[anchitj]

Need to have a check if partitionsCountPtr > 0

[anchitj]

Both could be merged in a single method too, something like

private ElectLeadersReport ExtractElectLeadersReport(IntPtr resultPtr)
{
    IntPtr topicPartitionsPtr = Librdkafka.ElectLeadersResult_partitions(resultPtr, out UIntPtr topicPartitionsCountPtr);
    

    IntPtr[] partitionsPtrArr = new IntPtr[(int)partitionsCountPtr];
    if ((int)partitionsCountPtr > 0)
    {
        Marshal.Copy(partitionsPtr, partitionsPtrArr, 0, (int)partitionsCountPtr);
    }

    ErrorCode reportErrorCode = ErrorCode.NoError;
    var topicPartitionErrors = partitionsPtrArr.Select(partitionPtr =>
    {
        var tpe = Marshal.PtrToStructure<rd_kafka_topic_partition_result>(partitionPtr);
        if (tpe.errCode != ErrorCode.NoError && reportErrorCode == ErrorCode.NoError)
        {
            reportErrorCode = tpe.errCode;
        }
        return new TopicPartitionError(
            tpe.topic,
            new Partition(tpe.partition),
            new Error(tpe.errCode, tpe.errorStr));
    }).ToList();

    return new ElectLeadersReport
    {
        ErrorCode = reportErrorCode,
        PartitionResults = topicPartitionErrors
    };
}

[anchitj]

Why are we making the user to pass ElectLeadersRequest instead of asking ElectionType and List<TopicPartition>? This is very much different from existing APIs

[anchitj]

Please also add an integration test

[anchitj]

Could rewrite something like this

        [Theory]
        [InlineData(-1)]
        [InlineData(0)]
        public async Task ElectLeadersAsync_InvalidRequestTimeout_ThrowsArgumentException(int timeoutSeconds)
        {
            using var adminClient = new AdminClientBuilder(GetTestConfig()).Build();
            var options = new ElectLeadersOptions { RequestTimeout = TimeSpan.FromSeconds(timeoutSeconds) };

            var exception = await Assert.ThrowsAsync<ArgumentException>(() =>
                adminClient.ElectLeadersAsync(
                    new ElectLeadersRequest { ElectionType = ElectionType.Preferred, Partitions = new List<TopicPartition> { new TopicPartition("topic", 0) } },
                    options)
            );

            Assert.Contains("RequestTimeout", exception.Message);
        }

        [Theory]
        [InlineData(ElectionType.Preferred)]
        [InlineData(ElectionType.Unclean)]
        public async Task ElectLeadersAsync_EmptyPartitions_ThrowsArgumentException(ElectionType electionType)
        {
            using var adminClient = new AdminClientBuilder(GetTestConfig()).Build();

            var exception = await Assert.ThrowsAsync<ArgumentException>(() =>
                adminClient.ElectLeadersAsync(
                    new ElectLeadersRequest { ElectionType = electionType, Partitions = new List<TopicPartition>() },
                    new ElectLeadersOptions())
            );

            Assert.Contains("No partitions specified", exception.Message);
        }

        [Theory]
        [InlineData(ElectionType.Preferred)]
        [InlineData(ElectionType.Unclean)]
        public async Task ElectLeadersAsync_DuplicatePartitions_ThrowsArgumentException(ElectionType electionType)
        {
            using var adminClient = new AdminClientBuilder(GetTestConfig()).Build();

            var exception = await Assert.ThrowsAsync<ArgumentException>(() =>
                adminClient.ElectLeadersAsync(
                    new ElectLeadersRequest
                    {
                        ElectionType = electionType,
                        Partitions = new List<TopicPartition> { new TopicPartition("topic", 0), new TopicPartition("topic", 0) }
                    },
                    new ElectLeadersOptions())
            );

            Assert.Contains("Duplicate partitions specified", exception.Message);
        }

        [Theory]
        [InlineData(ElectionType.Preferred)]
        [InlineData(ElectionType.Unclean)]
        [Timeout(1000)]
        public async Task ElectLeadersAsync_ValidRequest_TimesOut(ElectionType electionType)
        {
            using var adminClient = new AdminClientBuilder(GetTestConfig()).Build();

            var exception = await Assert.ThrowsAsync<KafkaException>(() =>
                adminClient.ElectLeadersAsync(
                    new ElectLeadersRequest
                    {
                        ElectionType = electionType,
                        Partitions = new List<TopicPartition> { new TopicPartition("topic", 0) }
                    },
                    new ElectLeadersOptions())
            );

            Assert.Contains("Local: Timed out", exception.Message);
        }

[anchitj]

brace at new line and spaces after if

[anchitj]

Typo in operation

[anchitj]

Rewrite as Options for the "AdminClient.ElectLeaders" method. to keep in sync with other classes

[anchitj]

Can we just return Error instead of ErrorCode? This would make sure the design aligns with other classes

[anchitj]

Why atleast one will be in error?

[anchitj]

Need to correct this

[anchitj]

formatting

[anchitj]

return type isn't being used so can be removed

[anchitj]

LGTM

[emasab]

Here are my proposed changes:

• rename to PartitionResults as TopicPartitions as in Go
• move ElectionType to Confluent.Kafka
• style fixes and a refactor in the example

[emasab]

Suggested change

	if ((args.Length -1 )%2 != 0)
	if ((args.Length -1 ) % 2 != 0)

[emasab]

Suggested change

	Console.WriteLine("usage: .. <bootstrapServers> elect-leaders electionType(0/1) <topic1> <partition1> ..");
	Console.WriteLine("usage: .. <bootstrapServers> elect-leaders <electionType> <topic1> <partition1> ..");

[emasab]

This can be common for the Result and Report by passing the partition list.

Suggested change

	static void PrintElectLeaderResults(ElectLeadersResult result)
	{
	Console.WriteLine(" ElectLeadersResult:");
	foreach (var partitionResult in result.PartitionResults)
	{
	Console.WriteLine($"Election successful in {partitionResult.Topic} {partitionResult.Partition}");
	}
	}
	static void PrintElectLeaderResults(List<TopicPartitionError> topicPartitions)
	{
	Console.WriteLine($"ElectLeaders response has {topicPartitions.Count} partition(s):");
	foreach (var partitionResult in topicPartitions)
	{
	if (!partitionResult.Error.IsError)
	Console.WriteLine($"Election successful in {partitionResult.Topic} {partitionResult.Partition}");
	else
	Console.WriteLine($"Election failed in {partitionResult.Topic} {partitionResult.Partition}: " +
	$"Code: {partitionResult.Error.Code}" +
	$", Reason: {partitionResult.Error.Reason}");
	}
	}

[emasab]

Suggested change

	Console.WriteLine("usage: .. <bootstrapServers> elect-leaders electionType(0/1) <topic1> <partition1> ..");
	Console.WriteLine("usage: .. <bootstrapServers> elect-leaders <electionType> <topic1> <partition1> ..");

[emasab]

Suggested change

	for (int i = 0; i < e.Results.PartitionResults.Count; ++i)
	{
	var result = e.Results.PartitionResults[i];
	if (!result.Error.IsError)
	{
	Console.WriteLine($"Elected leaders operation completed successfully for Topic {result.Topic} Partition {result.Partition}");
	}
	else
	{
	Console.WriteLine($"An error occurred in elect leaders operation in Topic {result.Topic} Partition {result.Partition}: Code: {result.Error.Code}" +
	$", Reason: {result.Error.Reason}");
	}
	}
	PrintElectLeaderResults(e.Results.TopicPartitions);

[emasab]

Add one newline

[emasab]

Suggested change

	{
	Error = new Error(ErrorCode.NoError),
	PartitionResults = new List<TopicPartitionError>()
	};
	return new ElectLeadersReport
	{
	TopicPartitions = new List<TopicPartitionError>()
	};

[PratRanj07]

It is esssential as it tells us to report no Error in case we dont recieve any topicPartitions in the result and we dont have to raise exception for this case in case of passing an empty list or for a null partitions case where no elections were required.

[emasab]

oh right this is still needed for checking that no error occurred

[emasab]

Suggested change

	var partitionResults = topicPartitionsPtrArr.Select(ptr =>
	var topicPartitions = topicPartitionsPtrArr.Select(ptr =>

[emasab]

Suggested change

	PartitionResults = partitionResults
	TopicPartitions = topicPartitions

[emasab]

Suggested change

	var result = new ElectLeadersResult(){
	PartitionResults = report.PartitionResults
	};
	var result = new ElectLeadersResult()
	{
	TopicPartitions = report.TopicPartitions
	};

[emasab]

LGTM! Thanks Pratyush!