Added a delete records api

examples/admin_delete_records/admin_delete_records.go

@@ -0,0 +1,92 @@
+/**
+ * Copyright 2022 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.
+ */
+
+// Delete Records before a particular offset in specified Topic Partition.
+
+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> "+
+				"<topic1> <partition1> <offset1> ...\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()
+
+	var topicPartitionOffsets []kafka.TopicPartition
+
+	argsCnt := len(os.Args)
+	i := 2
+	index := 0
+
+	for i < argsCnt {
+		if i+3 > argsCnt {
+			fmt.Printf("Expected %d arguments for partition %d, got %d\n", 3, index, argsCnt-i)
+			os.Exit(1)
+		}
+		topicName := os.Args[i]
+		partition, err := strconv.ParseInt(args[i+1], 10, 32)
+		if err != nil {
+			panic(err)
+		}
+		offset, err := strconv.ParseUint(args[i+2], 10, 64)
+		if err != nil {
+			panic(err)
+		}
+
+		topicPartitionOffsets = append(topicPartitionOffsets, kafka.TopicPartition{
+			Topic:     &topicName,
+			Partition: int32(partition),
+			Offset:    kafka.Offset(offset),
+		})
+		i += 3
+		index++
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
+	defer cancel()
+
+	res, err := ac.DeleteRecords(ctx, topicPartitionOffsets)
+	if err != nil {
+		fmt.Printf("Failed to delete records: %s\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Printf("Delete Records result: %+v\n", res)
+}

kafka/adminapi.go

@@ -258,6 +258,12 @@ type ListConsumerGroupsResult struct {
 	Errors []error
 }
 
+// DeleteRecordsResult represents the result of Delete Records call
+type DeleteRecordsResult struct {
+	// Slice of TopicPartition.
+	TopicPartitions []TopicPartition
+}
+
 // MemberAssignment represents the assignment of a consumer group member.
 type MemberAssignment struct {
 	// Partitions assigned to current member.
@@ -3397,6 +3403,82 @@ func (a *AdminClient) AlterUserScramCredentials(
 	return result, nil
 }
 
+// DeleteRecords delete all the records before the specified offset in
+// the given topic partition.
+//
+// Parameters:
+//   - `ctx` - context with the maximum amount of time to block, or nil for
+//     indefinite.
+//   - `topicpartitionoffsets` - Slice of TopicPartition where the deletion has to be performed. This should not be nil.
+//   - `options` - DeleteRecordsAdminOptions options.
+//
+// Returns DeleteRecordsResult, which contains a slice of
+// TopicPartitions containing the minimum offsets available in that
+// topic partition after the deletion operation has been performed or
+// any error if occured.
+func (a *AdminClient) DeleteRecords(ctx context.Context, topicpartitionoffsets []TopicPartition, options ...DeleteRecordsAdminOption) (result DeleteRecordsResult, err error) {
+
+	if topicpartitionoffsets == nil {
+		return result, newErrorFromString(ErrInvalidArg, "expected topicPartitionOffsets parameter.")
+	}
+	result = DeleteRecordsResult{}
+	err = a.verifyClient()
+	if err != nil {
+		return result, err
+	}
+
+	var delRecordCnt C.size_t = 1
+
+	// convert topicpartitionoffsets to rd_kafka_DeleteRecords_t** required by implementation
+	cTopicPartitionList := newCPartsFromTopicPartitions(topicpartitionoffsets)
+	defer C.rd_kafka_topic_partition_list_destroy(cTopicPartitionList)
+
+	cDelRecords := make([]*C.rd_kafka_DeleteRecords_t, delRecordCnt)
+	defer C.rd_kafka_DeleteRecords_destroy_array(&cDelRecords[0], C.size_t(delRecordCnt))
+
+	cDelRecords[0] = C.rd_kafka_DeleteRecords_new(cTopicPartitionList)
+
+	// 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_DELETERECORDS, 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_DeleteRecords (asynchronous).
+	C.rd_kafka_DeleteRecords(
+		a.handle.rk,
+		&cDelRecords[0],
+		C.size_t(delRecordCnt),
+		cOptions,
+		cQueue)
+
+	// Wait for result, error or context timeout.
+	rkev, err := a.waitResult(
+		ctx, cQueue, C.RD_KAFKA_EVENT_DELETERECORDS_RESULT)
+	if err != nil {
+		return result, err
+	}
+	defer C.rd_kafka_event_destroy(rkev)
+
+	cRes := C.rd_kafka_event_DeleteRecords_result(rkev)
+	cDeleteRecordsResultList := C.rd_kafka_DeleteRecords_result_offsets(cRes)
+
+	// Convert result from C to Go.
+	result.TopicPartitions = newTopicPartitionsFromCparts(cDeleteRecordsResultList)
+
+	return result, nil
+}
+
 // NewAdminClient creats a new AdminClient instance with a new underlying client instance
 func NewAdminClient(conf *ConfigMap) (*AdminClient, error) {
 

kafka/adminapi_test.go

@@ -881,6 +881,43 @@ func testAdminAPIsUserScramCredentials(what string, a *AdminClient, expDuration
 	}
 }
 
+func testAdminAPIsDeleteRecords(what string, a *AdminClient, expDuration time.Duration, t *testing.T) {
+	topic := "test"
+	partition := int32(0)
+	offset := Offset(2)
+	ctx, cancel := context.WithTimeout(context.Background(), expDuration)
+	defer cancel()
+	topicPartitionOffset := []TopicPartition{{Topic: &topic, Partition: partition, Offset: offset}}
+	_, err := a.DeleteRecords(ctx, topicPartitionOffset , 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.DeleteRecords(ctx, topicPartitionOffset, SetAdminRequestTimeout(-1))
+	if err == nil || err.(Error).Code() != ErrInvalidArg {
+		t.Fatalf("Expected ErrInvalidArg, not %v", err)
+	}
+
+	for _, options := range [][]DeleteRecordsAdminOption{
+		{},
+		{SetAdminRequestTimeout(time.Second)},
+	} {
+		// nil argument should fail, not being treated as empty
+		ctx, cancel = context.WithTimeout(context.Background(), expDuration)
+		defer cancel()
+		result, err := a.DeleteRecords(ctx, nil, options...)
+		if result.TopicPartitions != nil || err == nil {
+			t.Fatalf("Expected DeleteRecords 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)
 
@@ -1134,6 +1171,7 @@ func testAdminAPIs(what string, a *AdminClient, t *testing.T) {
 	testAdminAPIsListOffsets(what, a, expDuration, t)
 
 	testAdminAPIsUserScramCredentials(what, a, expDuration, t)
+	testAdminAPIsDeleteRecords(what, a, expDuration, t)
 }
 
 // TestAdminAPIs dry-tests most Admin APIs, no broker is needed.

kafka/adminoptions.go

@@ -143,6 +143,8 @@ func (ao AdminOptionRequestTimeout) supportsDescribeUserScramCredentials() {
 }
 func (ao AdminOptionRequestTimeout) supportsAlterUserScramCredentials() {
 }
+func (ao AdminOptionRequestTimeout) supportsDeleteRecords() {
+}
 func (ao AdminOptionRequestTimeout) apply(cOptions *C.rd_kafka_AdminOptions_t) error {
 	if !ao.isSet {
 		return nil
@@ -563,6 +565,14 @@ type ListOffsetsAdminOption interface {
 	apply(cOptions *C.rd_kafka_AdminOptions_t) error
 }
 
+// DeleteRecordsAdminOption - see setter.
+//
+// See SetAdminRequestTimeout.
+type DeleteRecordsAdminOption interface {
+  	supportsDeleteRecords()
+	apply(cOptions *C.rd_kafka_AdminOptions_t) error
+}	
+
 // AdminOption is a generic type not to be used directly.
 //
 // See CreateTopicsAdminOption et.al.

kafka/integration_test.go

@@ -3235,6 +3235,85 @@ func (its *IntegrationTestSuite) TestAdminClient_ListOffsets() {
 
 }
 
+// Test DeleteRecords API which deletes all the records before the specified offset
+// in the particular partition of the specified topic.
+func (its *IntegrationTestSuite) TestAdminClient_DeleteRecords() {
+    t := its.T()
+	bootstrapServers := testconf.Brokers
+	assert := its.Assert()
+
+	// Create a new AdminClient.
+	a := createAdminClient(t)
+	defer a.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Create a new topic to test the api and produce some messages to that topic
+	Topic := "test-delete"
+
+	topics := []TopicSpecification{TopicSpecification{Topic: Topic, NumPartitions: 1, ReplicationFactor: 1}}
+	createTopicResult, createTopicError := a.CreateTopics(ctx, topics)
+	assert.Nil(createTopicError, "Create Topics should not fail.")
+	assert.Equal(createTopicResult[0].Error.Code(), ErrNoError, "Create Topics Error Code should be ErrNoError.")
+
+    p, err := NewProducer(&ConfigMap{"bootstrap.servers": bootstrapServers})
+	assert.Nil(err, "Unable to create Producer.")
+	defer p.Close()
+
+	p.Produce(&Message{
+		TopicPartition: TopicPartition{Topic: &Topic, Partition: 0},
+		Value:          []byte("Message-1"),
+	}, nil)
+
+	p.Produce(&Message{
+		TopicPartition: TopicPartition{Topic: &Topic, Partition: 0},
+		Value:          []byte("Message-2"),
+	}, nil)
+
+	p.Produce(&Message{
+		TopicPartition: TopicPartition{Topic: &Topic, Partition: 0},
+		Value:          []byte("Message-3"),
+	}, nil)
+
+	p.Flush(5 * 1000)
+
+	// Delete the records till offset 2 in partion 0 of the topic
+	// The result will contain the minimum offset available after deletion in that Topic Partiton
+	var delRecordsTopicPartitionOffsets []TopicPartition
+	delRecordsTopicPartitionOffsets = append(delRecordsTopicPartitionOffsets, TopicPartition{
+		Topic:     &Topic,
+		Partition: int32(0),
+		Offset:    Offset(2),
+	})
+	var deleteRes DeleteRecordsResult
+	deleteRes, err = a.DeleteRecords(ctx,delRecordsTopicPartitionOffsets)
+	assert.Nil(err,"Delete Records should not fail")
+	var offsetAfterDeletion Offset
+	for _,tp := range deleteRes.TopicPartitions {
+		offsetAfterDeletion = tp.Offset
+	}
+
+	// Find the minimum uncommitted offset in that partition of the topic.
+	// It should be equal to the offset we get after the deletion operation
+	topicPartitionOffsets := make(map[TopicPartition]OffsetSpec)
+	tp1 := TopicPartition{Topic: &Topic, Partition: 0}
+	topicPartitionOffsets[tp1] = EarliestOffsetSpec
+	var results ListOffsetsResult
+	results, err = a.ListOffsets(ctx, topicPartitionOffsets, SetAdminIsolationLevel(IsolationLevelReadUncommitted))
+	assert.Nil(err, "ListOffsets should not fail.")
+
+	for _, info := range results.ResultInfos {
+		assert.Equal(info.Error.Code(), ErrNoError, "Error code should be ErrNoError.")
+		assert.Equal(info.Offset, offsetAfterDeletion, "Offset should be equal to the offset obtained after deleteion.")
+	}
+
+	delTopics := []string{Topic}
+	_, err = a.DeleteTopics(ctx, delTopics)
+	assert.Nil(err, "DeleteTopics should not fail.")
+
+}
+
 func TestIntegration(t *testing.T) {
 	its := new(IntegrationTestSuite)
 	testconfInit()