kafka golang 删除消息

在 Golang 中，可以使用 sarama 库来操作 Kafka。要删除特定偏移量之前的所有消息，可以使用 sarama 的 `OffsetDeleteRecordsRequest` API。

以下是一个简单的示例代码：

package main

import (
    "fmt"
    "github.com/Shopify/sarama"
)

func main() {
    broker := "localhost:9092"
    topic := "test"
    partition := int32(0)
    offset := int64(10)

    config := sarama.NewConfig()
    client, err := sarama.NewClient([]string{broker}, config)
    if err != nil {
        panic(err)
    }
    defer client.Close()

    admin, err := sarama.NewClusterAdminFromClient(client)
    if err != nil {
        panic(err)
    }
    defer admin.Close()

    req := &sarama.OffsetDeleteRecordsRequest{
        Topic: topic,
        PartitionOffsets: map[int32]*sarama.OffsetDeleteRequest{
            partition: {
                Offset: offset,
            },
        },
    }

    resp, err := admin.DeleteRecords(req)
    if err != nil {
        panic(err)
    }

    fmt.Printf("Deleted %d records from partition %d\n", resp.Responses[topic][partition].RecordsProcessed, partition)
}

在此示例中，我们首先创建一个 Kafka 客户端，然后使用 `sarama.NewClusterAdminFromClient` 创建一个 Kafka 管理员。接下来，我们创建一个 `OffsetDeleteRecordsRequest` 请求，并将要删除的偏移量指定为 `partition` 和 `offset` 参数。最后，我们使用 `admin.DeleteRecords` 方法来执行删除操作，并从响应中获取已处理的记录数。

请注意，如果要删除的偏移量不存在，或者已经被提交了，那么这个请求将会失败。因此，要确保在执行删除操作之前，消费者已经提交了偏移量。