add support of entity-default kafka quotas

Mongey · Jun 13, 2022 · 846e6a2 · 846e6a2
1 parent b77f249
commit 846e6a2
Show file tree

Hide file tree

Showing 4 changed files with 67 additions and 11 deletions.
diff --git a/kafka/kafka_acls.go b/kafka/kafka_acls.go
@@ -424,7 +424,7 @@ func (c *Client) ListACLs() ([]*sarama.ResourceAcls, error) {
 		if err != nil {
 			return nil, err
 		}
-		
+
 		log.Printf("[TRACE] ThrottleTime: %d", aclsR.ThrottleTime)
 
 		if err == nil {

diff --git a/kafka/kafka_quotas.go b/kafka/kafka_quotas.go
@@ -41,10 +41,19 @@ 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
@@ -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
@@ -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,
@@ -74,7 +74,7 @@ func quotaDelete(ctx context.Context, d *schema.ResourceData, meta interface{})
 func quotaRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	log.Println("[INFO] Reading Quota")
 	c := meta.(*LazyClient)
-	
+
 	entityType := d.Get("entity_type").(string)
 	entityName := d.Get("entity_name").(string)
 	log.Printf("[INFO] Reading Quota %s", entityName)

diff --git 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 {