From 135101e334d2d8556f1cd96377fdddeb799574eb Mon Sep 17 00:00:00 2001
From: azhurbilo <artemzhurbilo@gmail.com>
Date: Mon, 13 Jun 2022 02:02:40 +0300
Subject: [PATCH] add support of entity-default kafka quotas

---
 kafka/kafka_quotas.go              | 36 +++++++++++++++++++++-------
 kafka/resource_kafka_quota.go      |  2 +-
 kafka/resource_kafka_quota_test.go | 38 ++++++++++++++++++++++++++++++
 3 files changed, 66 insertions(+), 10 deletions(-)

diff --git a/kafka/kafka_quotas.go b/kafka/kafka_quotas.go
index e3e085d5..90e4dd8b 100644
--- a/kafka/kafka_quotas.go
+++ b/kafka/kafka_quotas.go
@@ -41,12 +41,21 @@ func (c *Client) AlterQuota(quota Quota, validateOnly bool) error {
 		return err
 	}
 
-	entity := sarama.QuotaEntityComponent{
-		EntityType: sarama.QuotaEntityType(quota.EntityType),
-		MatchType:  sarama.QuotaMatchExact,
-		Name:       quota.EntityName,
-	}
+	var entity sarama.QuotaEntityComponent
 
+	if quota.EntityName == "" {
+		entity = sarama.QuotaEntityComponent{
+			EntityType: sarama.QuotaEntityType(quota.EntityType),
+			MatchType:  sarama.QuotaMatchDefault,
+		}
+	} else {
+		entity = sarama.QuotaEntityComponent{
+			EntityType: sarama.QuotaEntityType(quota.EntityType),
+			MatchType:  sarama.QuotaMatchExact,
+			Name:       quota.EntityName,
+		}
+	}
+	
 	configs := quota.Ops
 
 	ops := []sarama.ClientQuotasOp{}
@@ -92,10 +101,19 @@ func (c *Client) DescribeQuota(entityType string, entityName string) (*Quota, er
 		return nil, err
 	}
 
-	entity := sarama.QuotaFilterComponent{
-		EntityType: sarama.QuotaEntityType(entityType),
-		MatchType:  sarama.QuotaMatchExact,
-		Match:      entityName,
+	var entity sarama.QuotaFilterComponent
+
+	if entityName == "" {
+		entity = sarama.QuotaFilterComponent{
+			EntityType: sarama.QuotaEntityType(entityType),
+			MatchType:  sarama.QuotaMatchDefault,
+		}
+	} else {
+		entity = sarama.QuotaFilterComponent{
+			EntityType: sarama.QuotaEntityType(entityType),
+			MatchType:  sarama.QuotaMatchExact,
+			Match:      entityName,
+		}
 	}
 
 	request := &sarama.DescribeClientQuotasRequest{
diff --git a/kafka/resource_kafka_quota.go b/kafka/resource_kafka_quota.go
index d6f13ec6..9b204d26 100644
--- a/kafka/resource_kafka_quota.go
+++ b/kafka/resource_kafka_quota.go
@@ -20,7 +20,7 @@ func kafkaQuotaResource() *schema.Resource {
 				Type:        schema.TypeString,
 				Required:    true,
 				ForceNew:    true,
-				Description: "The name of the entity",
+				Description: "The name of the entity (if entity_name is empty string, it will create default-entity Kafka quota)",
 			},
 			"entity_type": {
 				Type:             schema.TypeString,
diff --git a/kafka/resource_kafka_quota_test.go b/kafka/resource_kafka_quota_test.go
index a2b74a99..47b99e66 100644
--- a/kafka/resource_kafka_quota_test.go
+++ b/kafka/resource_kafka_quota_test.go
@@ -55,6 +55,44 @@ func TestAcc_QuotaConfigUpdate(t *testing.T) {
 	})
 }
 
+func TestAcc_DefaultEntityBasicQuota(t *testing.T) {
+	emptyEntityName := ""
+	bs := testBootstrapServers[0]
+
+	r.Test(t, r.TestCase{
+		ProviderFactories: overrideProviderFactory(),
+		PreCheck:          func() { testAccPreCheck(t) },
+		CheckDestroy:      testAccCheckQuotaDestroy,
+		Steps: []r.TestStep{
+			{
+				Config: cfgs(t, bs, fmt.Sprintf(testResourceQuota1, emptyEntityName, "4000000")),
+				Check:  testResourceQuota_initialCheck,
+			},
+		},
+	})
+}
+
+func TestAcc_DefaultEntityQuotaConfigUpdate(t *testing.T) {
+	emptyEntityName := ""
+	bs := testBootstrapServers[0]
+
+	r.Test(t, r.TestCase{
+		ProviderFactories: overrideProviderFactory(),
+		PreCheck:          func() { testAccPreCheck(t) },
+		CheckDestroy:      testAccCheckQuotaDestroy,
+		Steps: []r.TestStep{
+			{
+				Config: cfg(t, bs, fmt.Sprintf(testResourceQuota1, emptyEntityName, "4000000")),
+				Check:  testResourceQuota_initialCheck,
+			},
+			{
+				Config: cfg(t, bs, fmt.Sprintf(testResourceQuota1, emptyEntityName, "3000000")),
+				Check:  testResourceQuota_updateCheck,
+			},
+		},
+	})
+}
+
 func testResourceQuota_initialCheck(s *terraform.State) error {
 	resourceState := s.Modules[0].Resources["kafka_quota.test1"]
 	if resourceState == nil {