WIP

Author: Mongey

kafka/resource_kafka_topic_test.go

@@ -1,6 +1,7 @@
 package kafka
 
 import (
+	"errors"
 	"fmt"
 	"log"
 	"strconv"
@@ -144,8 +145,10 @@ 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)),
@@ -251,6 +254,8 @@ 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
@@ -269,12 +274,14 @@ func testResourceTopic_produceMessages(messages []*sarama.ProducerMessage) r.Tes
 
 		// 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 err
+						return errors.Join(produceErrs...)
 					}
 				}
 			} else {