add support of entity-default kafka quotas

README.md

@@ -184,20 +184,29 @@ provider "kafka" {
 }
 
 resource "kafka_quota" "test" {
- entity_name  = "client1"
- entity_type  = "client-id"
+ entity_name = "client1"
+ entity_type = "client-id"
  config = {
  "consumer_byte_rate" = "4000000"
  "producer_byte_rate" = "3500000"
  }
 }
+
+resource "kafka_quota" "default_user_quota" {
+ entity_name = ""
+ entity_type = "user"
+ config = {
+ "consumer_byte_rate" = "2000000"
+ "producer_byte_rate" = "1500000"
+ }
+}
 ```
 
 #### Properties
 
 | Property | Description |
 | -------------------- | ---------------------------------------------- |
-| `entity_name` | The name of the entity   |
+| `entity_name` | The name of the entity (if entity_name is empty string, it will create default-entity Kafka quota) |
 | `entity_type` | The entity type (client-id, user, ip) |
 | `config` | A map of string attributes for the entity |
 

examples/main.tf

@@ -51,10 +51,19 @@ resource "kafka_acl" "test" {
 }
 
 resource "kafka_quota" "quota1" {
- entity_name  = "client1"
- entity_type  = "client-id"
+ entity_name = "client1"
+ entity_type = "client-id"
  config = {
  "consumer_byte_rate" = "4000000"
  "producer_byte_rate" = "3500000"
  }
 }
+
+resource "kafka_quota" "default_user_quota" {
+ entity_name = ""
+ entity_type = "user"
+ config = {
+ "consumer_byte_rate" = "2000000"
+ "producer_byte_rate" = "1500000"
+ }
+}

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 {

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{

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)

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 {