Skip to content

Commit

Permalink
WIP
Browse files Browse the repository at this point in the history
  • Loading branch information
Mongey committed Aug 16, 2024
1 parent 0caf1d9 commit eae8e8d
Showing 1 changed file with 10 additions and 3 deletions.
13 changes: 10 additions & 3 deletions kafka/resource_kafka_topic_test.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package kafka

import (
"errors"
"fmt"
"log"
"strconv"
Expand Down Expand Up @@ -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)),
Expand Down Expand Up @@ -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
Expand All @@ -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 {
Expand Down

0 comments on commit eae8e8d

Please sign in to comment.