From 47febc02d2ca56a7ff978b777971bd33c6ef1548 Mon Sep 17 00:00:00 2001
From: mikatong <mikatong@tencent.com>
Date: Thu, 14 Jul 2022 19:55:23 +0800
Subject: [PATCH] fix: add kafka topic sweeper

---
 tencentcloud/resource_tc_ckafka_topic_test.go | 47 +++++++++++++++++++
 1 file changed, 47 insertions(+)

diff --git a/tencentcloud/resource_tc_ckafka_topic_test.go b/tencentcloud/resource_tc_ckafka_topic_test.go
index ebd34f6976..8f93301e62 100644
--- a/tencentcloud/resource_tc_ckafka_topic_test.go
+++ b/tencentcloud/resource_tc_ckafka_topic_test.go
@@ -3,13 +3,60 @@ package tencentcloud
 import (
 	"context"
 	"fmt"
+	"log"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/hashicorp/terraform-plugin-sdk/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/terraform"
 )
 
+func init() {
+	// go test -v ./tencentcloud -sweep=ap-guangzhou -sweep-run=tencentcloud_ckafka_topic
+	resource.AddTestSweepers("tencentcloud_ckafka_topic", &resource.Sweeper{
+		Name: "tencentcloud_ckafka_topic",
+		F: func(r string) error {
+			logId := getLogId(contextNil)
+			ctx := context.WithValue(context.TODO(), logIdKey, logId)
+			sharedClient, err := sharedClientForRegion(r)
+			if err != nil {
+				return fmt.Errorf("getting tencentcloud client error: %s", err.Error())
+			}
+			client := sharedClient.(*TencentCloudClient)
+			ckafkcService := CkafkaService{
+				client: client.apiV3Conn,
+			}
+			instanceId := defaultKafkaInstanceId
+			topicDetails, err := ckafkcService.DescribeCkafkaTopics(ctx, instanceId, "")
+			if err != nil {
+				return err
+			}
+			for _, topicDetail := range topicDetails {
+				log.Println(*topicDetail.TopicName)
+				topicName := *topicDetail.TopicName
+				now := time.Now()
+				createTime := time.Unix(*topicDetail.CreateTime, 0)
+				interval := now.Sub(createTime).Minutes()
+
+				if strings.HasPrefix(topicName, keepResource) || strings.HasPrefix(topicName, defaultResource) {
+					continue
+				}
+
+				if needProtect == 1 && int64(interval) < 30 {
+					continue
+				}
+				err := ckafkcService.DeleteCkafkaTopic(ctx, instanceId, topicName)
+				if err != nil {
+					return err
+				}
+			}
+
+			return nil
+		},
+	})
+}
+
 func TestAccTencentCloudCkafkaTopicResource(t *testing.T) {
 	t.Parallel()
 	resource.Test(t, resource.TestCase{