Merge branch 'master' into install-librdkafka-docs

jnadler · web-flow · commit d9ec726ff68c · 2019-11-15T16:11:21.000-08:00
diff --git a/Makefile b/Makefile
@@ -19,7 +19,7 @@ clean:
 format:
 	@go fmt
 	@go get golang.org/x/tools/cmd/goimports
-	@goimports -w $$(find . -name '*.go' | grep -v vendor)
+	@goimports -w $$(find . -name '*.go' | grep -v 'mock_*')
 
 lint: .makecache .makecache/lint
 
diff --git a/executor/message.go b/executor/message.go
@@ -68,11 +68,7 @@ type errorList struct {
 }
 
 func (el *errorList) addError(err error) {
-	if el.errors == nil {
-		el.errors = []error{err}
-	} else {
-		el.errors = append(el.errors, err)
-	}
+	el.errors = append(el.errors, err)
 }
 
 // deliverMessage recursively visits the source and all nodes and delivers the message if the node accepts this message type.
diff --git a/fbcontext/fbcontext.go b/fbcontext/fbcontext.go
@@ -89,9 +89,6 @@ func (c *ContextAware) Subscribe(messageTypes []string) {
 // AcceptsMessage is used when delivering messages to determine if a Source/Node requests delivery of messages with a given
 // messageType.
 func (c *ContextAware) AcceptsMessage(messageType string) bool {
-	if c.messageTypes == nil {
-		return false
-	}
 	for _, t := range c.messageTypes {
 		if messageType == t {
 			return true
diff --git a/go.mod b/go.mod
@@ -21,7 +21,7 @@ require (
 	github.com/tidwall/gjson v1.2.1 // indirect
 	github.com/tidwall/match v1.0.1 // indirect
 	github.com/tidwall/pretty v0.0.0-20180105212114-65a9db5fad51 // indirect
-	golang.org/x/sys v0.0.0-20191110163157-d32e6e3b99c4 // indirect
+	golang.org/x/sys v0.0.0-20191113165036-4c7a9d0fe056 // indirect
 	golang.org/x/time v0.0.0-20190308202827-9d24e82272b4
 	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 	gopkg.in/yaml.v2 v2.2.5
diff --git a/go.sum b/go.sum
@@ -175,15 +175,16 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894 h1:Cz4ceDQGXuKRnVBDTS23GTn/pU5OE2C0WrNTOYK1Uuc=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190801041406-cbf593c0f2f3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20191110163157-d32e6e3b99c4 h1:Hynbrlo6LbYI3H1IqXpkVDOcX/3HiPdhVEuyj5a59RM=
-golang.org/x/sys v0.0.0-20191110163157-d32e6e3b99c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191113165036-4c7a9d0fe056 h1:dHtDnRWQtSx0Hjq9kvKFpBh9uPPKfQN70NZZmvssGwk=
+golang.org/x/sys v0.0.0-20191113165036-4c7a9d0fe056/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
+golang.org/x/tools v0.0.0-20190312170243-e65039ee4138 h1:H3uGjxCR/6Ds0Mjgyp7LMK81+LvmbvWWEnJhzk1Pi9E=
 golang.org/x/tools v0.0.0-20190312170243-e65039ee4138/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 google.golang.org/api v0.3.1/go.mod h1:6wY9I6uQWHQ8EM57III9mq/AjF+i8G65rmVagqKMtkk=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
diff --git a/helpers.go b/helpers.go
@@ -5,6 +5,7 @@ import (
 	"strconv"
 )
 
+// Nodeconfig holds a Nodes configuration
 type Nodeconfig map[string]string
 
 // IntConfig validates and fetches the int-typed optional config value specified by 'name', using the 'defaultValue' if
diff --git a/internal/testnodetypes.go b/internal/testnodetypes.go
@@ -388,7 +388,7 @@ func (a *AsyncFilterNode) Receive(msg fbcontext.Message) error {
 	return errors.New("message not supported")
 }
 
-// IndexRequestBuildlerNode is a Node that converts strings to elasticsearch IndexRequests
+// IndexRequestBuilderNode is a Node that converts strings to elasticsearch IndexRequests
 type IndexRequestBuilderNode struct {
 	docNum int
 	fbcontext.ContextAware
diff --git a/leader/leader.go b/leader/leader.go
@@ -117,7 +117,7 @@ func (l *Leader) updateLeadership(status leaderelection.Status) {
 }
 
 // connectZk establishes a Zookeeper client connection.  The underlying client library does not block on creating the initial
-// conection, so no error will be returned from zk.Connect if zookeeper is not available.
+// connection, so no error will be returned from zk.Connect if zookeeper is not available.
 func connectZk(zkHosts string) (*zk.Conn, <-chan zk.Event) {
 	zks := strings.Split(zkHosts, ",")
 
diff --git a/message/kafkamessagesender.go b/message/kafkamessagesender.go
@@ -53,9 +53,9 @@ func (s KafkaMessageSender) Ack(msg Message) error {
 
 func (s *KafkaMessageSender) produceMessage(msg Message, ack bool) error {
 	wireMsg := &wireMessage{
-		Message:     msg,
-		Updated:     time.Now(),
-		Acknowleged: ack,
+		Message:      msg,
+		Updated:      time.Now(),
+		Acknowledged: ack,
 	}
 
 	wireMsgBytes, err := json.Marshal(wireMsg)
diff --git a/message/kafkamessagewire.go b/message/kafkamessagewire.go
@@ -6,9 +6,9 @@ import (
 
 // wireMessage is the structure that we send on the Kafka topic
 type wireMessage struct {
-	Message     Message   `json:"message"`
-	Updated     time.Time `json:"updated"`
-	Acknowleged bool      `json:"ack"`
+	Message      Message   `json:"message"`
+	Updated      time.Time `json:"updated"`
+	Acknowledged bool      `json:"ack"`
 }
 
 // uniqueKey concatenates messagetype with the key so that we don't risk collisions across message producers
diff --git a/message/kakfamessagereceiver.go b/message/kakfamessagereceiver.go
@@ -199,7 +199,7 @@ func (r *KafkaMessageReceiver) processMessage(value []byte) {
 	if !r.initialized {
 		r.initBuffer[uniqueKey(wireMsg.Message)] = wireMsg
 	} else {
-		if !wireMsg.Acknowleged {
+		if !wireMsg.Acknowledged {
 			r.deliverMessage(wireMsg.Message)
 		}
 	}
@@ -217,7 +217,7 @@ func (r *KafkaMessageReceiver) deliverMessage(msg Message) {
 func (r *KafkaMessageReceiver) processInitBuffer() {
 	log.WithField("buffered_messages", len(r.initBuffer)).Info("kafkamessagereceiver: delivering buffered messages")
 	for _, wireMsg := range r.initBuffer {
-		if !wireMsg.Acknowleged {
+		if !wireMsg.Acknowledged {
 			msg := wireMsg.Message
 			log.WithField("message_key", msg.Key).WithField("message_type", msg.MessageType).Infof("kafkamessagereceiver: delivering initial message")
 			r.deliverMessage(msg)
diff --git a/node/kafkaconsumer/consumermetrics_test.go b/node/kafkaconsumer/consumermetrics_test.go

Original file line number	Diff line number	Diff line change
`@@ -68,11 +68,7 @@ type errorList struct {`
`68`	`68`	`}`
`69`	`69`
`70`	`70`	`func (el *errorList) addError(err error) {`
`71`		`- if el.errors == nil {`
`72`		`- el.errors = []error{err}`
`73`		`- } else {`
`74`		`- el.errors = append(el.errors, err)`
`75`		`- }`
	`71`	`+ el.errors = append(el.errors, err)`
`76`	`72`	`}`
`77`	`73`
`78`	`74`	`// deliverMessage recursively visits the source and all nodes and delivers the message if the node accepts this message type.`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@ import (`
`5`	`5`	`"strconv"`
`6`	`6`	`)`
`7`	`7`
	`8`	`+// Nodeconfig holds a Nodes configuration`
`8`	`9`	`type Nodeconfig map[string]string`
`9`	`10`
`10`	`11`	`// IntConfig validates and fetches the int-typed optional config value specified by 'name', using the 'defaultValue' if`
Original file line number	Diff line number	Diff line change
`@@ -388,7 +388,7 @@ func (a *AsyncFilterNode) Receive(msg fbcontext.Message) error {`
`388`	`388`	`return errors.New("message not supported")`
`389`	`389`	`}`
`390`	`390`
`391`		`-// IndexRequestBuildlerNode is a Node that converts strings to elasticsearch IndexRequests`
	`391`	`+// IndexRequestBuilderNode is a Node that converts strings to elasticsearch IndexRequests`
`392`	`392`	`type IndexRequestBuilderNode struct {`
`393`	`393`	`docNum int`
`394`	`394`	`fbcontext.ContextAware`
Original file line number	Diff line number	Diff line change
`@@ -117,7 +117,7 @@ func (l *Leader) updateLeadership(status leaderelection.Status) {`
`117`	`117`	`}`
`118`	`118`
`119`	`119`	`// connectZk establishes a Zookeeper client connection. The underlying client library does not block on creating the initial`
`120`		`-// conection, so no error will be returned from zk.Connect if zookeeper is not available.`
	`120`	`+// connection, so no error will be returned from zk.Connect if zookeeper is not available.`
`121`	`121`	`func connectZk(zkHosts string) (*zk.Conn, <-chan zk.Event) {`
`122`	`122`	`zks := strings.Split(zkHosts, ",")`
`123`	`123`