SEBA-902 single-olt tests;
Pin protoc-gen-go to 1.3.2 to resolve compatibility issue;
Run go mod tidy / go mod vendor on importer;
Add Go Module support to demotest
Change-Id: Ifde824fc9a6317b0adc1e12bea54ee1f9b788906
diff --git a/vendor/github.com/Shopify/sarama/admin.go b/vendor/github.com/Shopify/sarama/admin.go
index 6c9b1e9..dd63484 100644
--- a/vendor/github.com/Shopify/sarama/admin.go
+++ b/vendor/github.com/Shopify/sarama/admin.go
@@ -2,8 +2,11 @@
import (
"errors"
+ "fmt"
"math/rand"
+ "strconv"
"sync"
+ "time"
)
// ClusterAdmin is the administrative client for Kafka, which supports managing and inspecting topics,
@@ -132,8 +135,45 @@
return ca.client.Controller()
}
-func (ca *clusterAdmin) CreateTopic(topic string, detail *TopicDetail, validateOnly bool) error {
+func (ca *clusterAdmin) refreshController() (*Broker, error) {
+ return ca.client.RefreshController()
+}
+// isErrNoController returns `true` if the given error type unwraps to an
+// `ErrNotController` response from Kafka
+func isErrNoController(err error) bool {
+ switch e := err.(type) {
+ case *TopicError:
+ return e.Err == ErrNotController
+ case *TopicPartitionError:
+ return e.Err == ErrNotController
+ case KError:
+ return e == ErrNotController
+ }
+ return false
+}
+
+// retryOnError will repeatedly call the given (error-returning) func in the
+// case that its response is non-nil and retriable (as determined by the
+// provided retriable func) up to the maximum number of tries permitted by
+// the admin client configuration
+func (ca *clusterAdmin) retryOnError(retriable func(error) bool, fn func() error) error {
+ var err error
+ for attempt := 0; attempt < ca.conf.Admin.Retry.Max; attempt++ {
+ err = fn()
+ if err == nil || !retriable(err) {
+ return err
+ }
+ Logger.Printf(
+ "admin/request retrying after %dms... (%d attempts remaining)\n",
+ ca.conf.Admin.Retry.Backoff/time.Millisecond, ca.conf.Admin.Retry.Max-attempt)
+ time.Sleep(ca.conf.Admin.Retry.Backoff)
+ continue
+ }
+ return err
+}
+
+func (ca *clusterAdmin) CreateTopic(topic string, detail *TopicDetail, validateOnly bool) error {
if topic == "" {
return ErrInvalidTopic
}
@@ -158,26 +198,31 @@
request.Version = 2
}
- b, err := ca.Controller()
- if err != nil {
- return err
- }
+ return ca.retryOnError(isErrNoController, func() error {
+ b, err := ca.Controller()
+ if err != nil {
+ return err
+ }
- rsp, err := b.CreateTopics(request)
- if err != nil {
- return err
- }
+ rsp, err := b.CreateTopics(request)
+ if err != nil {
+ return err
+ }
- topicErr, ok := rsp.TopicErrors[topic]
- if !ok {
- return ErrIncompleteResponse
- }
+ topicErr, ok := rsp.TopicErrors[topic]
+ if !ok {
+ return ErrIncompleteResponse
+ }
- if topicErr.Err != ErrNoError {
- return topicErr
- }
+ if topicErr.Err != ErrNoError {
+ if topicErr.Err == ErrNotController {
+ _, _ = ca.refreshController()
+ }
+ return topicErr
+ }
- return nil
+ return nil
+ })
}
func (ca *clusterAdmin) DescribeTopics(topics []string) (metadata []*TopicMetadata, err error) {
@@ -214,7 +259,7 @@
Topics: []string{},
}
- if ca.conf.Version.IsAtLeast(V0_11_0_0) {
+ if ca.conf.Version.IsAtLeast(V0_10_0_0) {
request.Version = 1
}
@@ -226,6 +271,16 @@
return response.Brokers, response.ControllerID, nil
}
+func (ca *clusterAdmin) findBroker(id int32) (*Broker, error) {
+ brokers := ca.client.Brokers()
+ for _, b := range brokers {
+ if b.ID() == id {
+ return b, nil
+ }
+ }
+ return nil, fmt.Errorf("could not find broker id %d", id)
+}
+
func (ca *clusterAdmin) findAnyBroker() (*Broker, error) {
brokers := ca.client.Brokers()
if len(brokers) > 0 {
@@ -308,7 +363,6 @@
}
func (ca *clusterAdmin) DeleteTopic(topic string) error {
-
if topic == "" {
return ErrInvalidTopic
}
@@ -322,25 +376,31 @@
request.Version = 1
}
- b, err := ca.Controller()
- if err != nil {
- return err
- }
+ return ca.retryOnError(isErrNoController, func() error {
+ b, err := ca.Controller()
+ if err != nil {
+ return err
+ }
- rsp, err := b.DeleteTopics(request)
- if err != nil {
- return err
- }
+ rsp, err := b.DeleteTopics(request)
+ if err != nil {
+ return err
+ }
- topicErr, ok := rsp.TopicErrorCodes[topic]
- if !ok {
- return ErrIncompleteResponse
- }
+ topicErr, ok := rsp.TopicErrorCodes[topic]
+ if !ok {
+ return ErrIncompleteResponse
+ }
- if topicErr != ErrNoError {
- return topicErr
- }
- return nil
+ if topicErr != ErrNoError {
+ if topicErr == ErrNotController {
+ _, _ = ca.refreshController()
+ }
+ return topicErr
+ }
+
+ return nil
+ })
}
func (ca *clusterAdmin) CreatePartitions(topic string, count int32, assignment [][]int32, validateOnly bool) error {
@@ -356,30 +416,34 @@
Timeout: ca.conf.Admin.Timeout,
}
- b, err := ca.Controller()
- if err != nil {
- return err
- }
+ return ca.retryOnError(isErrNoController, func() error {
+ b, err := ca.Controller()
+ if err != nil {
+ return err
+ }
- rsp, err := b.CreatePartitions(request)
- if err != nil {
- return err
- }
+ rsp, err := b.CreatePartitions(request)
+ if err != nil {
+ return err
+ }
- topicErr, ok := rsp.TopicPartitionErrors[topic]
- if !ok {
- return ErrIncompleteResponse
- }
+ topicErr, ok := rsp.TopicPartitionErrors[topic]
+ if !ok {
+ return ErrIncompleteResponse
+ }
- if topicErr.Err != ErrNoError {
- return topicErr
- }
+ if topicErr.Err != ErrNoError {
+ if topicErr.Err == ErrNotController {
+ _, _ = ca.refreshController()
+ }
+ return topicErr
+ }
- return nil
+ return nil
+ })
}
func (ca *clusterAdmin) DeleteRecords(topic string, partitionOffsets map[int32]int64) error {
-
if topic == "" {
return ErrInvalidTopic
}
@@ -432,8 +496,14 @@
return nil
}
-func (ca *clusterAdmin) DescribeConfig(resource ConfigResource) ([]ConfigEntry, error) {
+// Returns a bool indicating whether the resource request needs to go to a
+// specific broker
+func dependsOnSpecificNode(resource ConfigResource) bool {
+ return (resource.Type == BrokerResource && resource.Name != "") ||
+ resource.Type == BrokerLoggerResource
+}
+func (ca *clusterAdmin) DescribeConfig(resource ConfigResource) ([]ConfigEntry, error) {
var entries []ConfigEntry
var resources []*ConfigResource
resources = append(resources, &resource)
@@ -442,11 +512,31 @@
Resources: resources,
}
- b, err := ca.Controller()
+ if ca.conf.Version.IsAtLeast(V1_1_0_0) {
+ request.Version = 1
+ }
+
+ if ca.conf.Version.IsAtLeast(V2_0_0_0) {
+ request.Version = 2
+ }
+
+ var (
+ b *Broker
+ err error
+ )
+
+ // DescribeConfig of broker/broker logger must be sent to the broker in question
+ if dependsOnSpecificNode(resource) {
+ id, _ := strconv.Atoi(resource.Name)
+ b, err = ca.findBroker(int32(id))
+ } else {
+ b, err = ca.findAnyBroker()
+ }
if err != nil {
return nil, err
}
+ _ = b.Open(ca.client.Config())
rsp, err := b.DescribeConfigs(request)
if err != nil {
return nil, err
@@ -466,7 +556,6 @@
}
func (ca *clusterAdmin) AlterConfig(resourceType ConfigResourceType, name string, entries map[string]*string, validateOnly bool) error {
-
var resources []*AlterConfigsResource
resources = append(resources, &AlterConfigsResource{
Type: resourceType,
@@ -479,11 +568,23 @@
ValidateOnly: validateOnly,
}
- b, err := ca.Controller()
+ var (
+ b *Broker
+ err error
+ )
+
+ // AlterConfig of broker/broker logger must be sent to the broker in question
+ if dependsOnSpecificNode(ConfigResource{Name: name, Type: resourceType}) {
+ id, _ := strconv.Atoi(name)
+ b, err = ca.findBroker(int32(id))
+ } else {
+ b, err = ca.findAnyBroker()
+ }
if err != nil {
return err
}
+ _ = b.Open(ca.client.Config())
rsp, err := b.AlterConfigs(request)
if err != nil {
return err
@@ -518,7 +619,6 @@
}
func (ca *clusterAdmin) ListAcls(filter AclFilter) ([]ResourceAcls, error) {
-
request := &DescribeAclsRequest{AclFilter: filter}
if ca.conf.Version.IsAtLeast(V2_0_0_0) {
@@ -566,7 +666,6 @@
for _, mACL := range fr.MatchingAcls {
mAcls = append(mAcls, *mACL)
}
-
}
return mAcls, nil
}
@@ -580,7 +679,6 @@
return nil, err
}
groupsPerBroker[controller] = append(groupsPerBroker[controller], group)
-
}
for broker, brokerGroups := range groupsPerBroker {
@@ -623,7 +721,6 @@
}
groupMaps <- groups
-
}(b, ca.conf)
}