increase producer rety backoff in acc test

kafka/client.go

@@ -57,17 +57,17 @@
 
 func NewClient(config *Config) (*Client, error) {
 	if config == nil {
-		return nil, errors.New("Cannot create client without kafka config")
+		return nil, errors.New("cannot create client without kafka config")
 	}
 
 	log.Printf("[TRACE] configuring bootstrap_servers %v", config.copyWithMaskedSensitiveValues())
 	if config.BootstrapServers == nil {
-		return nil, fmt.Errorf("No bootstrap_servers provided")
+		return nil, fmt.Errorf("no bootstrap_servers provided")
 	}
 
 	bootstrapServers := *(config.BootstrapServers)
 	if bootstrapServers == nil {
-		return nil, fmt.Errorf("No bootstrap_servers provided")
+		return nil, fmt.Errorf("no bootstrap_servers provided")
 	}
 
 	log.Printf("[INFO] configuring kafka client with %v", config.copyWithMaskedSensitiveValues())
@@ -369,7 +369,9 @@
 		return err
 	}
 
-	return admin.AlterPartitionReassignments(t.Name, *assignment)
+	err = admin.AlterPartitionReassignments(t.Name, *assignment)
+	c.client.RefreshMetadata(t.Name)
+	return err
 }
 
 func (c *Client) buildAssignment(t Topic) (*[][]int32, error) {

kafka/config.go

@@ -282,7 +282,7 @@ func newTLSConfig(clientCert, clientKey, caCert, clientKeyPassphrase string) (*t
 	ok := caCertPool.AppendCertsFromPEM(caBytes)
 	log.Printf("[TRACE] set cert pool %v", ok)
 	if !ok {
-		return &tlsConfig, fmt.Errorf("Couldn't add the caPem")
+		return &tlsConfig, fmt.Errorf("couldn't add the caPem")
 	}
 
 	tlsConfig.RootCAs = caCertPool

kafka/data_source_kafka_topic.go

@@ -43,13 +43,12 @@ func dataSourceTopicRead(d *schema.ResourceData, meta interface{}) error {
 
 	client := meta.(*LazyClient)
 	topic, err := client.ReadTopic(name, true)
-
 	if err != nil {
 		log.Printf("[ERROR] Error getting topic %s from Kafka: %s", name, err)
 		_, ok := err.(TopicMissingError)
 
 		if ok {
-			return fmt.Errorf("Could not find topic '%s'", name)
+			return fmt.Errorf("could not find topic '%s'", name)
 		}
 
 		return err

kafka/kafka_acls.go

@@ -36,19 +36,19 @@ func tfToAclCreation(s StringlyTypedACL) (*sarama.AclCreation, error) {
 
 	op := stringToOperation(s.ACL.Operation)
 	if op == unknownConversion {
-		return acl, fmt.Errorf("Unknown operation: %s", s.ACL.Operation)
+		return acl, fmt.Errorf("unknown operation: %s", s.ACL.Operation)
 	}
 	pType := stringToAclPermissionType(s.ACL.PermissionType)
 	if pType == unknownConversion {
-		return acl, fmt.Errorf("Unknown permission type: %s", s.ACL.PermissionType)
+		return acl, fmt.Errorf("unknown permission type: %s", s.ACL.PermissionType)
 	}
 	rType := stringToACLResource(s.Resource.Type)
 	if rType == unknownConversion {
-		return acl, fmt.Errorf("Unknown resource type: %s", s.Resource.Type)
+		return acl, fmt.Errorf("unknown resource type: %s", s.Resource.Type)
 	}
 	patternType := stringToACLPrefix(s.Resource.PatternTypeFilter)
 	if patternType == unknownConversion {
-		return acl, fmt.Errorf("Unknown pattern type filter: '%s'", s.Resource.PatternTypeFilter)
+		return acl, fmt.Errorf("unknown pattern type filter: '%s'", s.Resource.PatternTypeFilter)
 	}
 
 	acl.Acl = sarama.Acl{
@@ -77,25 +77,25 @@ func tfToAclFilter(s StringlyTypedACL) (sarama.AclFilter, error) {
 
 	op := stringToOperation(s.ACL.Operation)
 	if op == unknownConversion {
-		return f, fmt.Errorf("Unknown operation: %s", s.ACL.Operation)
+		return f, fmt.Errorf("unknown operation: %s", s.ACL.Operation)
 	}
 	f.Operation = op
 
 	pType := stringToAclPermissionType(s.ACL.PermissionType)
 	if pType == unknownConversion {
-		return f, fmt.Errorf("Unknown permission type: %s", s.ACL.PermissionType)
+		return f, fmt.Errorf("unknown permission type: %s", s.ACL.PermissionType)
 	}
 	f.PermissionType = pType
 
 	rType := stringToACLResource(s.Resource.Type)
 	if rType == unknownConversion {
-		return f, fmt.Errorf("Unknown resource type: %s", s.Resource.Type)
+		return f, fmt.Errorf("unknown resource type: %s", s.Resource.Type)
 	}
 	f.ResourceType = rType
 
 	patternType := stringToACLPrefix(s.Resource.PatternTypeFilter)
 	if patternType == unknownConversion {
-		return f, fmt.Errorf("Unknown pattern type filter: '%s'", s.Resource.PatternTypeFilter)
+		return f, fmt.Errorf("unknown pattern type filter: '%s'", s.Resource.PatternTypeFilter)
 	}
 	f.ResourcePatternTypeFilter = patternType
 

kafka/kafka_quotas.go

@@ -113,7 +113,7 @@ func (c *Client) DescribeQuota(entityType string, entityName string) (*Quota, er
 
 	if err == nil {
 		if quotaR.ErrorCode != sarama.ErrNoError {
-			return nil, fmt.Errorf("Error describing quota %s", quotaR.ErrorCode)
+			return nil, fmt.Errorf("error describing quota %s", quotaR.ErrorCode)
 		}
 	}
 

kafka/kafka_user_scram_credentials.go

@@ -74,7 +74,7 @@ func (c *Client) DescribeUserScramCredential(username string, mechanism string)
 
 	num := len(results)
 	if num != 1 {
-		return nil, fmt.Errorf("Got %d results (expected 1) when describing user scram credential %s", num, username)
+		return nil, fmt.Errorf("got %d results (expected 1) when describing user scram credential %s", num, username)
 	}
 
 	res := results[0]

kafka/migrate_kafka_acl_state.go

@@ -13,7 +13,7 @@ func migrateKafkaAclState(v int, is *terraform.InstanceState, meta interface{})
 		log.Println("[INFO] Found Kafka ACL v0 state; migrating to v1")
 		return migrateKafkaAclV0toV1(is)
 	default:
-		return is, fmt.Errorf("Unexpected schema version: %d", v)
+		return is, fmt.Errorf("unexpected schema version: %d", v)
 	}
 }
 

kafka/resource_kafka_quota_test.go

@@ -24,7 +24,7 @@ func TestAcc_BasicQuota(t *testing.T) {
 		CheckDestroy:      testAccCheckQuotaDestroy,
 		Steps: []r.TestStep{
 			{
-				Config: cfgs(t, bs, fmt.Sprintf(testResourceQuota1, quotaEntityName, "4000000")),
+				Config: cfgs(bs, fmt.Sprintf(testResourceQuota1, quotaEntityName, "4000000")),
 				Check:  testResourceQuota_initialCheck,
 			},
 		},
@@ -174,7 +174,7 @@ func testAccCheckQuotaDestroy(s *terraform.State) error {
 }
 
 // lintignore:AT004
-func cfgs(t *testing.T, bs string, extraCfg string) string {
+func cfgs(bs string, extraCfg string) string {
 	return fmt.Sprintf(`
 provider "kafka" {
 	bootstrap_servers = ["%s"]

kafka/resource_kafka_topic.go

@@ -221,7 +221,7 @@ func topicDelete(ctx context.Context, d *schema.ResourceData, meta interface{})
 	stateConf := &retry.StateChangeConf{
 		Pending:      []string{"Pending"},
 		Target:       []string{"Deleted"},
-		Refresh:      topicDeleteFunc(c, d.Id(), t),
+		Refresh:      topicDeleteFunc(c, t),
 		Timeout:      300 * time.Second,
 		Delay:        3 * time.Second,
 		PollInterval: 2 * time.Second,
@@ -237,7 +237,7 @@ func topicDelete(ctx context.Context, d *schema.ResourceData, meta interface{})
 	return nil
 }
 
-func topicDeleteFunc(client *LazyClient, id string, t Topic) retry.StateRefreshFunc {
+func topicDeleteFunc(client *LazyClient, t Topic) retry.StateRefreshFunc {
 	return func() (result interface{}, s string, err error) {
 		topic, err := client.ReadTopic(t.Name, true)
 

kafka/resource_kafka_topic_test.go

@@ -1,6 +1,7 @@
 package kafka
 
 import (
+	"errors"
 	"fmt"
 	"log"
 	"strconv"
@@ -144,14 +145,17 @@ func TestAcc_TopicAlterReplicationFactor(t *testing.T) {
 
 	r.Test(t, r.TestCase{
 		ProviderFactories: overrideProviderFactory(),
-		PreCheck:          func() { testAccPreCheck(t) },
-		CheckDestroy:      testAccCheckTopicDestroy,
+		PreCheck: func() {
+			testAccPreCheck(t)
+		},
+		CheckDestroy: testAccCheckTopicDestroy,
 		Steps: []r.TestStep{
 			{
 				Config: cfg(t, bs, fmt.Sprintf(testResourceTopic_updateRF, topicName, 1, 7)),
 				Check: r.ComposeTestCheckFunc(
+					testResourceTopic_initialCheck,
 					testResourceTopic_produceMessages(messages),
-					testResourceTopic_initialCheck),
+				),
 			},
 			{
 				Config: cfg(t, bs, fmt.Sprintf(testResourceTopic_updateRF, topicName, 3, 7)),
@@ -250,8 +254,12 @@ func testResourceTopic_produceMessages(messages []*sarama.ProducerMessage) r.Tes
 		}
 		kafkaConfig.Producer.Return.Errors = true
 		kafkaConfig.Producer.Return.Successes = true
+		kafkaConfig.Metadata.Full = true
+
 		kafkaConfig.Producer.RequiredAcks = sarama.WaitForAll
 		kafkaConfig.Producer.Timeout = 90 * time.Second
+		kafkaConfig.Producer.Retry.Max = 5
+		kafkaConfig.Producer.Retry.Backoff = time.Second
 
 		producer, err := sarama.NewSyncProducer(*config.BootstrapServers, kafkaConfig)
 		if err != nil {
@@ -264,8 +272,21 @@ func testResourceTopic_produceMessages(messages []*sarama.ProducerMessage) r.Tes
 			}
 		}()
 
-		if err := producer.SendMessages(messages); err != nil {
-			return err
+		// rety 5 times
+		retries := 5
+		produceErrs := make([]error, 0, retries)
+		for i := 0; i < retries; i++ {
+			if errs := producer.SendMessages(messages); errs != nil {
+				produceErrs = append(produceErrs, errs)
+				for _, err := range errs.(sarama.ProducerErrors) {
+					log.Println("[ERROR] Write to kafka failed: ", err)
+					if i == retries-1 {
+						return errors.Join(produceErrs...)
+					}
+				}
+			} else {
+				return nil
+			}
 		}
 
 		return nil

kafka/resource_kafka_user_scram_credential.go

@@ -69,7 +69,7 @@ func importSCRAM(ctx context.Context, d *schema.ResourceData, m interface{}) ([]
 			return nil, errSet.err
 		}
 	} else {
-		return nil, fmt.Errorf("Failed importing resource; expected format is username|scram_mechanism|password - got %v segments instead of 3", len(parts))
+		return nil, fmt.Errorf("failed importing resource; expected format is username|scram_mechanism|password - got %v segments instead of 3", len(parts))
 	}
 
 	return []*schema.ResourceData{d}, nil

kafka/resource_kafka_user_scram_credential_test.go

@@ -27,7 +27,7 @@ func TestAcc_UserScramCredentialBasic(t *testing.T) {
 		CheckDestroy:      testAccCheckUserScramCredentialDestroy,
 		Steps: []r.TestStep{
 			{
-				Config: cfgs(t, bs, fmt.Sprintf(testResourceUserScramCredential_SHA256, username)),
+				Config: cfgs(bs, fmt.Sprintf(testResourceUserScramCredential_SHA256, username)),
 				Check:  testResourceUserScramCredentialCheck_withoutIterations,
 			},
 		},

kafka/scram_client.go

@@ -8,8 +8,10 @@ import (
 	"github.com/xdg/scram"
 )
 
-var SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }
-var SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }
+var (
+	SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }
+	SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }
+)
 
 type XDGSCRAMClient struct {
 	*scram.Client

kafka/topic.go

@@ -33,17 +33,16 @@ func ReplicaCount(c sarama.Client, topic string, partitions []int32) (int, error
 	for _, p := range partitions {
 		replicas, err := c.Replicas(topic, p)
 		if err != nil {
-			return -1, errors.New("Could not get replicas for partition")
+			return -1, errors.New("could not get replicas for partition")
 		}
 		if count == -1 {
 			count = len(replicas)
 		}
 		if count != len(replicas) {
-			return count, fmt.Errorf("The replica count isn't the same across partitions %d != %d", count, len(replicas))
+			return count, fmt.Errorf("replica count isn't the same across partitions %d != %d", count, len(replicas))
 		}
 	}
 	return count, nil
-
 }
 
 func configToResources(topic Topic) []*sarama.AlterConfigsResource {