ElectLeader API implemented

examples/.gitignore

@@ -9,6 +9,7 @@ admin_delete_records/admin_delete_records
 admin_delete_topics/admin_delete_topics
 admin_describe_acls/admin_describe_acls
 admin_describe_config/admin_describe_config
+admin_elect_leaders/admin_elect_leaders
 admin_incremental_alter_configs/admin_incremental_alter_configs
 admin_describe_consumer_groups/admin_describe_consumer_groups
 admin_list_consumer_groups/admin_list_consumer_groups

examples/README.md

@@ -28,6 +28,8 @@ Examples
 
   [admin_describe_topics](admin_describe_topics) - Describe topics
 
+  [admin_elect_leaders](admin_elect_leaders) - Perform Preferred/ Unclean elections for the specified Topic Partitions
+
   [admin_list_consumer_group_offsets](admin_list_consumer_group_offsets) - List consumer group offsets
 
   [admin_list_offsets](admin_list_offsets) - List partition offsets

examples/admin_elect_leaders/admin_elect_leaders.go

@@ -0,0 +1,96 @@
+/**
+ * Copyright 2024 Confluent Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Perform Preffered or Unclean Elections for the specified Topic Partitions.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strconv"
+	"time"
+
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+)
+
+func main() {
+	args := os.Args
+
+	if len(args) < 5 {
+		fmt.Fprintf(os.Stderr,
+			"Usage: %s <bootstrap_servers> "+
+				"<election_type (Preferred/Unclean)> <topic1> <partition1> ...\n",
+			args[0])
+		os.Exit(1)
+	}
+
+	// Create new AdminClient.
+	ac, err := kafka.NewAdminClient(&kafka.ConfigMap{
+		"bootstrap.servers": args[1],
+	})
+	if err != nil {
+		fmt.Printf("Failed to create Admin client: %s\n", err)
+		os.Exit(1)
+	}
+	defer ac.Close()
+
+	electionType, err := kafka.ElectionTypeFromString(args[2])
+	if err != nil {
+		fmt.Printf("Invalid election type: %s\n", err)
+		os.Exit(1)
+	}
+
+	var topicPartitionList []kafka.TopicPartition
+
+	argsCnt := len(os.Args)
+	i := 3
+	index := 0
+
+	for i < argsCnt {
+		if i+2 > argsCnt {
+			fmt.Printf("Expected %d arguments for partition %d, got %d\n", 2, index, argsCnt-i)
+			os.Exit(1)
+		}
+		topicName := os.Args[i]
+		partition, err := strconv.ParseInt(args[i+1], 10, 32)
+		if err != nil {
+			panic(err)
+		}
+
+		topicPartitionList = append(topicPartitionList, kafka.TopicPartition{
+			Topic:     &topicName,
+			Partition: int32(partition),
+		})
+		i += 2
+		index++
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
+	defer cancel()
+
+	res, err := ac.ElectLeaders(ctx, kafka.NewElectLeaderRequest(electionType, topicPartitionList))
+	if err != nil {
+		kafkaErr, ok := err.(kafka.Error) // Type assertion
+		if ok && kafkaErr.Code() != kafka.ErrNoError {
+			fmt.Printf("Failed to elect Leaders: %s\n", err)
+			os.Exit(1)
+		}
+	}
+
+	fmt.Printf("ElectLeaders result: %+v\n", res)
+}

kafka/adminapi.go

@@ -1033,6 +1033,50 @@ type ListOffsetsResult struct {
 	ResultInfos map[TopicPartition]ListOffsetsResultInfo
 }
 
+// ElectionType represents the type of election to be performed
+type ElectionType int
+
+const (
+	// ElectionTypePreferred - Preferred election type
+	ElectionTypePreferred ElectionType = C.RD_KAFKA_ELECTION_TYPE_PREFERRED
+	// ElectionTypeUnclean - Unclean election type
+	ElectionTypeUnclean ElectionType = C.RD_KAFKA_ELECTION_TYPE_UNCLEAN
+)
+
+func ElectionTypeFromString(electionTypeString string) (ElectionType, error) {
+	switch strings.ToUpper(electionTypeString) {
+	case "PREFERRED":
+		return ElectionTypePreferred, nil
+	case "UNCLEAN":
+		return ElectionTypeUnclean, nil
+	default:
+		return ElectionTypePreferred, NewError(ErrInvalidArg, "Unknown election type", false)
+	}
+}
+
+// ElectLeaderRequest holds parameters for the type of election to be performed and
+// the topic partitions for which election has to be performed
+type ElectLeaderRequest struct {
+	// Election type to be performed
+	electionType ElectionType
+	// TopicPartitions for which election has to be performed
+	partitions []TopicPartition
+}
+
+func NewElectLeaderRequest(electionType ElectionType, partitions []TopicPartition) ElectLeaderRequest {
+	return ElectLeaderRequest{
+		electionType: electionType,
+		partitions:   partitions,
+	}
+}
+
+// ElectLeaderResult holds the result of the election performed
+type ElectLeaderResult struct {
+	// TopicPartitions for which election has been performed and specific error if any
+	// that occured while election in the specific TopicPartition.
+	topicPartitions []TopicPartition
+}
+
 // waitResult waits for a result event on cQueue or the ctx to be cancelled, whichever happens
 // first.
 // The returned result event is checked for errors its error is returned if set.
@@ -1513,6 +1557,25 @@ func (a *AdminClient) cConfigResourceToResult(cRes **C.rd_kafka_ConfigResource_t
 	return result, nil
 }
 
+// Convert a C rd_kafka_topic_partition_result_t array to a Go TopicPartition list.
+func newTopicPartitionsFromCTopicPartitionResult(cResponse **C.rd_kafka_topic_partition_result_t, size C.size_t) (partitions []TopicPartition) {
+
+	partCnt := int(size)
+
+	partitions = make([]TopicPartition, partCnt)
+
+	for i := 0; i < partCnt; i++ {
+		topic := C.GoString(C.rd_kafka_topic_partition_result_topic(C.rd_kafka_ElectionResult_partition_by_idx(cResponse, C.size_t(i))))
+		partitions[i].Topic = &topic
+		partitions[i].Partition = int32(C.rd_kafka_topic_partition_result_partition(C.rd_kafka_ElectionResult_partition_by_idx(cResponse, C.size_t(i))))
+		cErrorCode := C.rd_kafka_topic_partition_result_error(C.rd_kafka_ElectionResult_partition_by_idx(cResponse, C.size_t(i)))
+		cErrorStr := C.rd_kafka_topic_partition_result_error_string(C.rd_kafka_ElectionResult_partition_by_idx(cResponse, C.size_t(i)))
+		partitions[i].Error = newErrorFromCString(cErrorCode, cErrorStr)
+	}
+
+	return partitions
+}
+
 // ClusterID returns the cluster ID as reported in broker metadata.
 //
 // Note on cancellation: Although the underlying C function respects the
@@ -3482,6 +3545,79 @@ func (a *AdminClient) DeleteRecords(ctx context.Context, recordsToDelete []Topic
 	return result, nil
 }
 
+// ElectLeaders performs Preferred or Unclean Elections for the specified topic Partitions.
+// 
+// Parameters:
+//   - `ctx` - context with the maximum amount of time to block, or nil for
+//     indefinite.
+//   - `electLeaderRequest` - ElectLeaderRequest containing the election type
+//         and the partitions to elect leaders for.
+//   - `options` - ElectLeadersAdminOption options.
+//
+// Returns ElectLeaderResult, which contains a slice of TopicPartitions containing the partitions for which the leader election was performed.
+// User has to check all the elements of the result to check any error occured per partition.
+// err returns any top level level that occured during the operation.
+
+func (a *AdminClient) ElectLeaders(ctx context.Context, electLeaderRequest ElectLeaderRequest, options ...ElectLeadersAdminOption) (result ElectLeaderResult, err error) {
+
+	if len(electLeaderRequest.partitions) == 0 {
+		return result, newErrorFromString(ErrInvalidArg, "expected non-empty partitions")
+	}
+
+	result = ElectLeaderResult{}
+
+	err = a.verifyClient()
+	if err != nil {
+		return result, err
+	}
+
+	cTopicPartitions := newCPartsFromTopicPartitions(electLeaderRequest.partitions)
+	defer C.rd_kafka_topic_partition_list_destroy(cTopicPartitions)
+
+	cElectLeaderRequest := C.rd_kafka_ElectLeader_new(C.rd_kafka_ElectionType_t(electLeaderRequest.electionType), cTopicPartitions)
+	defer C.rd_kafka_ElectLeader_destroy(cElectLeaderRequest)
+
+	// Convert Go AdminOptions (if any) to C AdminOptions.
+	genericOptions := make([]AdminOption, len(options))
+	for i := range options {
+		genericOptions[i] = options[i]
+	}
+	cOptions, err := adminOptionsSetup(
+		a.handle, C.RD_KAFKA_ADMIN_OP_ELECTLEADER, genericOptions)
+	if err != nil {
+		return result, err
+	}
+	defer C.rd_kafka_AdminOptions_destroy(cOptions)
+
+	// Create temporary queue for async operation.
+	cQueue := C.rd_kafka_queue_new(a.handle.rk)
+	defer C.rd_kafka_queue_destroy(cQueue)
+
+	// Call rd_kafka_ElectLeader (asynchronous).
+	C.rd_kafka_ElectLeader(
+		a.handle.rk,
+		cElectLeaderRequest,
+		cOptions,
+		cQueue)
+
+	// Wait for result, error or context timeout.
+	rkev, err := a.waitResult(
+		ctx, cQueue, C.RD_KAFKA_EVENT_ELECTLEADER_RESULT)
+	if err != nil {
+		return result, err
+	}
+	defer C.rd_kafka_event_destroy(rkev)
+
+	cRes := C.rd_kafka_event_ElectLeader_result(rkev)
+	err = newError(C.rd_kafka_ElectionResult_error(cRes))
+
+	var cResponseSize C.size_t
+	cResultPartitions := C.rd_kafka_ElectionResult_partition(cRes, &cResponseSize)
+	result.topicPartitions = newTopicPartitionsFromCTopicPartitionResult(cResultPartitions, cResponseSize)
+
+	return result, err
+}
+
 // NewAdminClient creats a new AdminClient instance with a new underlying client instance
 func NewAdminClient(conf *ConfigMap) (*AdminClient, error) {
 

kafka/adminapi_test.go

@@ -918,6 +918,80 @@ func testAdminAPIsDeleteRecords(what string, a *AdminClient, expDuration time.Du
 	}
 }
 
+func testAdminAPIsElectLeaders(what string, a *AdminClient, expDuration time.Duration, t *testing.T) {
+	topic := "test"
+	partition := int32(0)
+	ctx, cancel := context.WithTimeout(context.Background(), expDuration)
+	defer cancel()
+	topicPartition := []TopicPartition{{Topic: &topic, Partition: partition}}
+	emptyTopicPartition := []TopicPartition{}
+	electLeaderRequestPreferred := ElectLeaderRequest{
+		electionType: ElectionTypePreferred,
+		partitions: topicPartition,
+	}
+	electLeadersRequestUnclean := ElectLeaderRequest{
+		electionType: ElectionTypeUnclean,
+		partitions:   topicPartition,
+	}
+
+	_, err := a.ElectLeaders(ctx, electLeaderRequestPreferred, SetAdminRequestTimeout(time.Second))
+	if err == nil || ctx.Err() != context.DeadlineExceeded {
+		t.Fatalf("Expected context deadline exceeded, got %s and %s\n",
+			err, ctx.Err())
+	}
+
+	_, err = a.ElectLeaders(ctx, electLeadersRequestUnclean, SetAdminRequestTimeout(time.Second))
+	if err == nil || ctx.Err() != context.DeadlineExceeded {
+		t.Fatalf("Expected context deadline exceeded, got %s and %s\n",
+			err, ctx.Err())
+	}
+
+	// Invalid option value
+	_, err = a.ElectLeaders(ctx, electLeaderRequestPreferred, SetAdminRequestTimeout(-1))
+	if err == nil || err.(Error).Code() != ErrInvalidArg {
+		t.Fatalf("Expected ErrInvalidArg, not %v", err)
+	}
+
+	_, err = a.ElectLeaders(ctx, electLeadersRequestUnclean, SetAdminRequestTimeout(-1))
+	if err == nil || err.(Error).Code() != ErrInvalidArg {
+		t.Fatalf("Expected ErrInvalidArg, not %v", err)
+	}
+
+	for _, options := range [][]ElectLeadersAdminOption{
+		{},
+		{SetAdminRequestTimeout(time.Second)},
+	} {
+		// empty list should fail
+		ctx, cancel = context.WithTimeout(context.Background(), expDuration)
+		defer cancel()
+		emptyElectLeaderRequestPreferred := ElectLeaderRequest{
+			electionType: ElectionTypePreferred,
+			partitions:   emptyTopicPartition,
+		}
+		emptyElectLeaderRequestUnclean := ElectLeaderRequest{
+			electionType: ElectionTypeUnclean,
+			partitions:   emptyTopicPartition,
+		}
+		result, err := a.ElectLeaders(ctx, emptyElectLeaderRequestPreferred, options...)
+		if result.topicPartitions != nil || err == nil {
+			t.Fatalf("Expected ElectLeaders to fail, but got result: %v, err: %v",
+				result, err)
+		}
+		if err.(Error).Code() != ErrInvalidArg {
+			t.Fatalf("Expected ErrInvalidArg, not %v", err)
+		}
+
+		result, err = a.ElectLeaders(ctx, emptyElectLeaderRequestUnclean, options...)
+		if result.topicPartitions != nil || err == nil {
+			t.Fatalf("Expected ElectLeaders to fail, but got result: %v, err: %v",
+				result, err)
+		}
+		if err.(Error).Code() != ErrInvalidArg {
+			t.Fatalf("Expected ErrInvalidArg, not %v", err)
+		}
+	}
+}
+
 func testAdminAPIs(what string, a *AdminClient, t *testing.T) {
 	t.Logf("AdminClient API testing on %s: %s", a, what)
 
@@ -1172,6 +1246,7 @@ func testAdminAPIs(what string, a *AdminClient, t *testing.T) {
 
 	testAdminAPIsUserScramCredentials(what, a, expDuration, t)
 	testAdminAPIsDeleteRecords(what, a, expDuration, t)
+	testAdminAPIsElectLeaders(what, a, expDuration, t)
 }
 
 // TestAdminAPIs dry-tests most Admin APIs, no broker is needed.

kafka/adminoptions.go

@@ -53,6 +53,8 @@ func (ao AdminOptionOperationTimeout) supportsCreatePartitions() {
 }
 func (ao AdminOptionOperationTimeout) supportsDeleteRecords() {
 }
+func (ao AdminOptionOperationTimeout) supportsElectLeaders() {
+}
 
 func (ao AdminOptionOperationTimeout) apply(cOptions *C.rd_kafka_AdminOptions_t) error {
 	if !ao.isSet {
@@ -147,6 +149,8 @@ func (ao AdminOptionRequestTimeout) supportsAlterUserScramCredentials() {
 }
 func (ao AdminOptionRequestTimeout) supportsDeleteRecords() {
 }
+func (ao AdminOptionRequestTimeout) supportsElectLeaders() {
+}
 func (ao AdminOptionRequestTimeout) apply(cOptions *C.rd_kafka_AdminOptions_t) error {
 	if !ao.isSet {
 		return nil
@@ -575,6 +579,14 @@ type DeleteRecordsAdminOption interface {
 	apply(cOptions *C.rd_kafka_AdminOptions_t) error
 }
 
+// ElectLeadersAdminOption - see setter.
+//
+// See SetAdminRequestTimeout, SetAdminOperationTimeout.
+type ElectLeadersAdminOption interface {
+	supportsElectLeaders()
+	apply(cOptions *C.rd_kafka_AdminOptions_t) error
+}
+
 // AdminOption is a generic type not to be used directly.
 //
 // See CreateTopicsAdminOption et.al.