updates vendor dep

vendor/github.com/Shopify/sarama/LICENSE

+Copyright (c) 2013 Evan Huus
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

vendor/github.com/Shopify/sarama/Makefile

+default: fmt vet errcheck test
+
+# Taken from https://github.com/codecov/example-go#caveat-multiple-files
+test:
+	echo "" > coverage.txt
+	for d in `go list ./... | grep -v vendor`; do \
+		go test -v -timeout 60s -race -coverprofile=profile.out -covermode=atomic $$d; \
+		if [ -f profile.out ]; then \
+			cat profile.out >> coverage.txt; \
+			rm profile.out; \
+		fi \
+	done
+
+vet:
+	go vet ./...
+
+errcheck:
+	errcheck github.com/Shopify/sarama/...
+
+fmt:
+	@if [ -n "$$(go fmt ./...)" ]; then echo 'Please run go fmt on your code.' && exit 1; fi
+
+install_dependencies: install_errcheck get
+
+install_errcheck:
+	go get github.com/kisielk/errcheck
+
+get:
+	go get -t

vendor/github.com/Shopify/sarama/README.md

+sarama
+======
+
+[![GoDoc](https://godoc.org/github.com/Shopify/sarama?status.png)](https://godoc.org/github.com/Shopify/sarama)
+[![Build Status](https://travis-ci.org/Shopify/sarama.svg?branch=master)](https://travis-ci.org/Shopify/sarama)
+[![Coverage](https://codecov.io/gh/Shopify/sarama/branch/master/graph/badge.svg)](https://codecov.io/gh/Shopify/sarama)
+
+Sarama is an MIT-licensed Go client library for [Apache Kafka](https://kafka.apache.org/) version 0.8 (and later).
+
+### Getting started
+
+- API documentation and examples are available via [godoc](https://godoc.org/github.com/Shopify/sarama).
+- Mocks for testing are available in the [mocks](./mocks) subpackage.
+- The [examples](./examples) directory contains more elaborate example applications.
+- The [tools](./tools) directory contains command line tools that can be useful for testing, diagnostics, and instrumentation.
+
+You might also want to look at the [Frequently Asked Questions](https://github.com/Shopify/sarama/wiki/Frequently-Asked-Questions).
+
+### Compatibility and API stability
+
+Sarama provides a "2 releases + 2 months" compatibility guarantee: we support
+the two latest stable releases of Kafka and Go, and we provide a two month
+grace period for older releases. This means we currently officially support
+Go 1.9 and 1.8, and Kafka 1.0 through 0.10, although older releases are
+still likely to work.
+
+Sarama follows semantic versioning and provides API stability via the gopkg.in service.
+You can import a version with a guaranteed stable API via http://gopkg.in/Shopify/sarama.v1.
+A changelog is available [here](CHANGELOG.md).
+
+### Contributing
+
+* Get started by checking our [contribution guidelines](https://github.com/Shopify/sarama/blob/master/.github/CONTRIBUTING.md).
+* Read the [Sarama wiki](https://github.com/Shopify/sarama/wiki) for more
+  technical and design details.
+* The [Kafka Protocol Specification](https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol)
+  contains a wealth of useful information.
+* For more general issues, there is [a google group](https://groups.google.com/forum/#!forum/kafka-clients) for Kafka client developers.
+* If you have any questions, just ask!

vendor/github.com/Shopify/sarama/Vagrantfile

+# -*- mode: ruby -*-
+# vi: set ft=ruby :
+
+# Vagrantfile API/syntax version. Don't touch unless you know what you're doing!
+VAGRANTFILE_API_VERSION = "2"
+
+# We have 5 * 192MB ZK processes and 5 * 320MB Kafka processes => 2560MB
+MEMORY = 3072
+
+Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
+  config.vm.box = "ubuntu/trusty64"
+
+  config.vm.provision :shell, path: "vagrant/provision.sh"
+
+  config.vm.network "private_network", ip: "192.168.100.67"
+
+  config.vm.provider "virtualbox" do |v|
+    v.memory = MEMORY
+  end
+end

vendor/github.com/Shopify/sarama/acl_bindings.go

+package sarama
+
+type Resource struct {
+	ResourceType AclResourceType
+	ResourceName string
+}
+
+func (r *Resource) encode(pe packetEncoder) error {
+	pe.putInt8(int8(r.ResourceType))
+
+	if err := pe.putString(r.ResourceName); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (r *Resource) decode(pd packetDecoder, version int16) (err error) {
+	resourceType, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	r.ResourceType = AclResourceType(resourceType)
+
+	if r.ResourceName, err = pd.getString(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+type Acl struct {
+	Principal      string
+	Host           string
+	Operation      AclOperation
+	PermissionType AclPermissionType
+}
+
+func (a *Acl) encode(pe packetEncoder) error {
+	if err := pe.putString(a.Principal); err != nil {
+		return err
+	}
+
+	if err := pe.putString(a.Host); err != nil {
+		return err
+	}
+
+	pe.putInt8(int8(a.Operation))
+	pe.putInt8(int8(a.PermissionType))
+
+	return nil
+}
+
+func (a *Acl) decode(pd packetDecoder, version int16) (err error) {
+	if a.Principal, err = pd.getString(); err != nil {
+		return err
+	}
+
+	if a.Host, err = pd.getString(); err != nil {
+		return err
+	}
+
+	operation, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.Operation = AclOperation(operation)
+
+	permissionType, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.PermissionType = AclPermissionType(permissionType)
+
+	return nil
+}
+
+type ResourceAcls struct {
+	Resource
+	Acls []*Acl
+}
+
+func (r *ResourceAcls) encode(pe packetEncoder) error {
+	if err := r.Resource.encode(pe); err != nil {
+		return err
+	}
+
+	if err := pe.putArrayLength(len(r.Acls)); err != nil {
+		return err
+	}
+	for _, acl := range r.Acls {
+		if err := acl.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (r *ResourceAcls) decode(pd packetDecoder, version int16) error {
+	if err := r.Resource.decode(pd, version); err != nil {
+		return err
+	}
+
+	n, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	r.Acls = make([]*Acl, n)
+	for i := 0; i < n; i++ {
+		r.Acls[i] = new(Acl)
+		if err := r.Acls[i].decode(pd, version); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

vendor/github.com/Shopify/sarama/acl_create_request.go

+package sarama
+
+type CreateAclsRequest struct {
+	AclCreations []*AclCreation
+}
+
+func (c *CreateAclsRequest) encode(pe packetEncoder) error {
+	if err := pe.putArrayLength(len(c.AclCreations)); err != nil {
+		return err
+	}
+
+	for _, aclCreation := range c.AclCreations {
+		if err := aclCreation.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *CreateAclsRequest) decode(pd packetDecoder, version int16) (err error) {
+	n, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	c.AclCreations = make([]*AclCreation, n)
+
+	for i := 0; i < n; i++ {
+		c.AclCreations[i] = new(AclCreation)
+		if err := c.AclCreations[i].decode(pd, version); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (d *CreateAclsRequest) key() int16 {
+	return 30
+}
+
+func (d *CreateAclsRequest) version() int16 {
+	return 0
+}
+
+func (d *CreateAclsRequest) requiredVersion() KafkaVersion {
+	return V0_11_0_0
+}
+
+type AclCreation struct {
+	Resource
+	Acl
+}
+
+func (a *AclCreation) encode(pe packetEncoder) error {
+	if err := a.Resource.encode(pe); err != nil {
+		return err
+	}
+	if err := a.Acl.encode(pe); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (a *AclCreation) decode(pd packetDecoder, version int16) (err error) {
+	if err := a.Resource.decode(pd, version); err != nil {
+		return err
+	}
+	if err := a.Acl.decode(pd, version); err != nil {
+		return err
+	}
+
+	return nil
+}

vendor/github.com/Shopify/sarama/acl_create_request_test.go

+package sarama
+
+import "testing"
+
+var (
+	aclCreateRequest = []byte{
+		0, 0, 0, 1,
+		3, // resource type = group
+		0, 5, 'g', 'r', 'o', 'u', 'p',
+		0, 9, 'p', 'r', 'i', 'n', 'c', 'i', 'p', 'a', 'l',
+		0, 4, 'h', 'o', 's', 't',
+		2, // all
+		2, // deny
+	}
+)
+
+func TestCreateAclsRequest(t *testing.T) {
+	req := &CreateAclsRequest{
+		AclCreations: []*AclCreation{{
+			Resource: Resource{
+				ResourceType: AclResourceGroup,
+				ResourceName: "group",
+			},
+			Acl: Acl{
+				Principal:      "principal",
+				Host:           "host",
+				Operation:      AclOperationAll,
+				PermissionType: AclPermissionDeny,
+			}},
+		},
+	}
+
+	testRequest(t, "create request", req, aclCreateRequest)
+}

vendor/github.com/Shopify/sarama/acl_create_response.go

+package sarama
+
+import "time"
+
+type CreateAclsResponse struct {
+	ThrottleTime         time.Duration
+	AclCreationResponses []*AclCreationResponse
+}
+
+func (c *CreateAclsResponse) encode(pe packetEncoder) error {
+	pe.putInt32(int32(c.ThrottleTime / time.Millisecond))
+
+	if err := pe.putArrayLength(len(c.AclCreationResponses)); err != nil {
+		return err
+	}
+
+	for _, aclCreationResponse := range c.AclCreationResponses {
+		if err := aclCreationResponse.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *CreateAclsResponse) decode(pd packetDecoder, version int16) (err error) {
+	throttleTime, err := pd.getInt32()
+	if err != nil {
+		return err
+	}
+	c.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
+
+	n, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	c.AclCreationResponses = make([]*AclCreationResponse, n)
+	for i := 0; i < n; i++ {
+		c.AclCreationResponses[i] = new(AclCreationResponse)
+		if err := c.AclCreationResponses[i].decode(pd, version); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (d *CreateAclsResponse) key() int16 {
+	return 30
+}
+
+func (d *CreateAclsResponse) version() int16 {
+	return 0
+}
+
+func (d *CreateAclsResponse) requiredVersion() KafkaVersion {
+	return V0_11_0_0
+}
+
+type AclCreationResponse struct {
+	Err    KError
+	ErrMsg *string
+}
+
+func (a *AclCreationResponse) encode(pe packetEncoder) error {
+	pe.putInt16(int16(a.Err))
+
+	if err := pe.putNullableString(a.ErrMsg); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (a *AclCreationResponse) decode(pd packetDecoder, version int16) (err error) {
+	kerr, err := pd.getInt16()
+	if err != nil {
+		return err
+	}
+	a.Err = KError(kerr)
+
+	if a.ErrMsg, err = pd.getNullableString(); err != nil {
+		return err
+	}
+
+	return nil
+}

vendor/github.com/Shopify/sarama/acl_create_response_test.go

+package sarama
+
+import (
+	"testing"
+	"time"
+)
+
+var (
+	createResponseWithError = []byte{
+		0, 0, 0, 100,
+		0, 0, 0, 1,
+		0, 42,
+		0, 5, 'e', 'r', 'r', 'o', 'r',
+	}
+
+	createResponseArray = []byte{
+		0, 0, 0, 100,
+		0, 0, 0, 2,
+		0, 42,
+		0, 5, 'e', 'r', 'r', 'o', 'r',
+		0, 0,
+		255, 255,
+	}
+)
+
+func TestCreateAclsResponse(t *testing.T) {
+	errmsg := "error"
+	resp := &CreateAclsResponse{
+		ThrottleTime: 100 * time.Millisecond,
+		AclCreationResponses: []*AclCreationResponse{{
+			Err:    ErrInvalidRequest,
+			ErrMsg: &errmsg,
+		}},
+	}
+
+	testResponse(t, "response with error", resp, createResponseWithError)
+
+	resp.AclCreationResponses = append(resp.AclCreationResponses, new(AclCreationResponse))
+
+	testResponse(t, "response array", resp, createResponseArray)
+}

vendor/github.com/Shopify/sarama/acl_delete_request.go

+package sarama
+
+type DeleteAclsRequest struct {
+	Filters []*AclFilter
+}
+
+func (d *DeleteAclsRequest) encode(pe packetEncoder) error {
+	if err := pe.putArrayLength(len(d.Filters)); err != nil {
+		return err
+	}
+
+	for _, filter := range d.Filters {
+		if err := filter.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (d *DeleteAclsRequest) decode(pd packetDecoder, version int16) (err error) {
+	n, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	d.Filters = make([]*AclFilter, n)
+	for i := 0; i < n; i++ {
+		d.Filters[i] = new(AclFilter)
+		if err := d.Filters[i].decode(pd, version); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (d *DeleteAclsRequest) key() int16 {
+	return 31
+}
+
+func (d *DeleteAclsRequest) version() int16 {
+	return 0
+}
+
+func (d *DeleteAclsRequest) requiredVersion() KafkaVersion {
+	return V0_11_0_0
+}

vendor/github.com/Shopify/sarama/acl_delete_request_test.go

+package sarama
+
+import "testing"
+
+var (
+	aclDeleteRequestNulls = []byte{
+		0, 0, 0, 1,
+		1,
+		255, 255,
+		255, 255,
+		255, 255,
+		11,
+		3,
+	}
+
+	aclDeleteRequest = []byte{
+		0, 0, 0, 1,
+		1, // any
+		0, 6, 'f', 'i', 'l', 't', 'e', 'r',
+		0, 9, 'p', 'r', 'i', 'n', 'c', 'i', 'p', 'a', 'l',
+		0, 4, 'h', 'o', 's', 't',
+		4, // write
+		3, // allow
+	}
+
+	aclDeleteRequestArray = []byte{
+		0, 0, 0, 2,
+		1,
+		0, 6, 'f', 'i', 'l', 't', 'e', 'r',
+		0, 9, 'p', 'r', 'i', 'n', 'c', 'i', 'p', 'a', 'l',
+		0, 4, 'h', 'o', 's', 't',
+		4, // write
+		3, // allow
+		2,
+		0, 5, 't', 'o', 'p', 'i', 'c',
+		255, 255,
+		255, 255,
+		6,
+		2,
+	}
+)
+
+func TestDeleteAclsRequest(t *testing.T) {
+	req := &DeleteAclsRequest{
+		Filters: []*AclFilter{{
+			ResourceType:   AclResourceAny,
+			Operation:      AclOperationAlterConfigs,
+			PermissionType: AclPermissionAllow,
+		}},
+	}
+
+	testRequest(t, "delete request nulls", req, aclDeleteRequestNulls)
+
+	req.Filters[0].ResourceName = nullString("filter")
+	req.Filters[0].Principal = nullString("principal")
+	req.Filters[0].Host = nullString("host")
+	req.Filters[0].Operation = AclOperationWrite
+
+	testRequest(t, "delete request", req, aclDeleteRequest)
+
+	req.Filters = append(req.Filters, &AclFilter{
+		ResourceType:   AclResourceTopic,
+		ResourceName:   nullString("topic"),
+		Operation:      AclOperationDelete,
+		PermissionType: AclPermissionDeny,
+	})
+
+	testRequest(t, "delete request array", req, aclDeleteRequestArray)
+}

vendor/github.com/Shopify/sarama/acl_delete_response.go

+package sarama
+
+import "time"
+
+type DeleteAclsResponse struct {
+	ThrottleTime    time.Duration
+	FilterResponses []*FilterResponse
+}
+
+func (a *DeleteAclsResponse) encode(pe packetEncoder) error {
+	pe.putInt32(int32(a.ThrottleTime / time.Millisecond))
+
+	if err := pe.putArrayLength(len(a.FilterResponses)); err != nil {
+		return err
+	}
+
+	for _, filterResponse := range a.FilterResponses {
+		if err := filterResponse.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (a *DeleteAclsResponse) decode(pd packetDecoder, version int16) (err error) {
+	throttleTime, err := pd.getInt32()
+	if err != nil {
+		return err
+	}
+	a.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
+
+	n, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+	a.FilterResponses = make([]*FilterResponse, n)
+
+	for i := 0; i < n; i++ {
+		a.FilterResponses[i] = new(FilterResponse)
+		if err := a.FilterResponses[i].decode(pd, version); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (d *DeleteAclsResponse) key() int16 {
+	return 31
+}
+
+func (d *DeleteAclsResponse) version() int16 {
+	return 0
+}
+
+func (d *DeleteAclsResponse) requiredVersion() KafkaVersion {
+	return V0_11_0_0
+}
+
+type FilterResponse struct {
+	Err          KError
+	ErrMsg       *string
+	MatchingAcls []*MatchingAcl
+}
+
+func (f *FilterResponse) encode(pe packetEncoder) error {
+	pe.putInt16(int16(f.Err))
+	if err := pe.putNullableString(f.ErrMsg); err != nil {
+		return err
+	}
+
+	if err := pe.putArrayLength(len(f.MatchingAcls)); err != nil {
+		return err
+	}
+	for _, matchingAcl := range f.MatchingAcls {
+		if err := matchingAcl.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (f *FilterResponse) decode(pd packetDecoder, version int16) (err error) {
+	kerr, err := pd.getInt16()
+	if err != nil {
+		return err
+	}
+	f.Err = KError(kerr)
+
+	if f.ErrMsg, err = pd.getNullableString(); err != nil {
+		return err
+	}
+
+	n, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+	f.MatchingAcls = make([]*MatchingAcl, n)
+	for i := 0; i < n; i++ {
+		f.MatchingAcls[i] = new(MatchingAcl)
+		if err := f.MatchingAcls[i].decode(pd, version); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+type MatchingAcl struct {
+	Err    KError
+	ErrMsg *string
+	Resource
+	Acl
+}
+
+func (m *MatchingAcl) encode(pe packetEncoder) error {
+	pe.putInt16(int16(m.Err))
+	if err := pe.putNullableString(m.ErrMsg); err != nil {
+		return err
+	}
+
+	if err := m.Resource.encode(pe); err != nil {
+		return err
+	}
+
+	if err := m.Acl.encode(pe); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (m *MatchingAcl) decode(pd packetDecoder, version int16) (err error) {
+	kerr, err := pd.getInt16()
+	if err != nil {
+		return err
+	}
+	m.Err = KError(kerr)
+
+	if m.ErrMsg, err = pd.getNullableString(); err != nil {
+		return err
+	}
+
+	if err := m.Resource.decode(pd, version); err != nil {
+		return err
+	}
+
+	if err := m.Acl.decode(pd, version); err != nil {
+		return err
+	}
+
+	return nil
+}

vendor/github.com/Shopify/sarama/acl_delete_response_test.go

+package sarama
+
+import (
+	"testing"
+	"time"
+)
+
+var (
+	deleteAclsResponse = []byte{
+		0, 0, 0, 100,
+		0, 0, 0, 1,
+		0, 0, // no error
+		255, 255, // no error message
+		0, 0, 0, 1, // 1 matching acl
+		0, 0, // no error
+		255, 255, // no error message
+		2, // resource type
+		0, 5, 't', 'o', 'p', 'i', 'c',
+		0, 9, 'p', 'r', 'i', 'n', 'c', 'i', 'p', 'a', 'l',
+		0, 4, 'h', 'o', 's', 't',
+		4,
+		3,
+	}
+)
+
+func TestDeleteAclsResponse(t *testing.T) {
+	resp := &DeleteAclsResponse{
+		ThrottleTime: 100 * time.Millisecond,
+		FilterResponses: []*FilterResponse{{
+			MatchingAcls: []*MatchingAcl{{
+				Resource: Resource{ResourceType: AclResourceTopic, ResourceName: "topic"},
+				Acl:      Acl{Principal: "principal", Host: "host", Operation: AclOperationWrite, PermissionType: AclPermissionAllow},
+			}},
+		}},
+	}
+
+	testResponse(t, "", resp, deleteAclsResponse)
+}

vendor/github.com/Shopify/sarama/acl_describe_request.go

+package sarama
+
+type DescribeAclsRequest struct {
+	AclFilter
+}
+
+func (d *DescribeAclsRequest) encode(pe packetEncoder) error {
+	return d.AclFilter.encode(pe)
+}
+
+func (d *DescribeAclsRequest) decode(pd packetDecoder, version int16) (err error) {
+	return d.AclFilter.decode(pd, version)
+}
+
+func (d *DescribeAclsRequest) key() int16 {
+	return 29
+}
+
+func (d *DescribeAclsRequest) version() int16 {
+	return 0
+}
+
+func (d *DescribeAclsRequest) requiredVersion() KafkaVersion {
+	return V0_11_0_0
+}

vendor/github.com/Shopify/sarama/acl_describe_request_test.go

+package sarama
+
+import (
+	"testing"
+)
+
+var (
+	aclDescribeRequest = []byte{
+		2, // resource type
+		0, 5, 't', 'o', 'p', 'i', 'c',
+		0, 9, 'p', 'r', 'i', 'n', 'c', 'i', 'p', 'a', 'l',
+		0, 4, 'h', 'o', 's', 't',
+		5, // acl operation
+		3, // acl permission type
+	}
+)
+
+func TestAclDescribeRequest(t *testing.T) {
+	resourcename := "topic"
+	principal := "principal"
+	host := "host"
+
+	req := &DescribeAclsRequest{
+		AclFilter{
+			ResourceType:   AclResourceTopic,
+			ResourceName:   &resourcename,
+			Principal:      &principal,
+			Host:           &host,
+			Operation:      AclOperationCreate,
+			PermissionType: AclPermissionAllow,
+		},
+	}
+
+	testRequest(t, "", req, aclDescribeRequest)
+}

vendor/github.com/Shopify/sarama/acl_describe_response.go

+package sarama
+
+import "time"
+
+type DescribeAclsResponse struct {
+	ThrottleTime time.Duration
+	Err          KError
+	ErrMsg       *string
+	ResourceAcls []*ResourceAcls
+}
+
+func (d *DescribeAclsResponse) encode(pe packetEncoder) error {
+	pe.putInt32(int32(d.ThrottleTime / time.Millisecond))
+	pe.putInt16(int16(d.Err))
+
+	if err := pe.putNullableString(d.ErrMsg); err != nil {
+		return err
+	}
+
+	if err := pe.putArrayLength(len(d.ResourceAcls)); err != nil {
+		return err
+	}
+
+	for _, resourceAcl := range d.ResourceAcls {
+		if err := resourceAcl.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (d *DescribeAclsResponse) decode(pd packetDecoder, version int16) (err error) {
+	throttleTime, err := pd.getInt32()
+	if err != nil {
+		return err
+	}
+	d.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
+
+	kerr, err := pd.getInt16()
+	if err != nil {
+		return err
+	}
+	d.Err = KError(kerr)
+
+	errmsg, err := pd.getString()
+	if err != nil {
+		return err
+	}
+	if errmsg != "" {
+		d.ErrMsg = &errmsg
+	}
+
+	n, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+	d.ResourceAcls = make([]*ResourceAcls, n)
+
+	for i := 0; i < n; i++ {
+		d.ResourceAcls[i] = new(ResourceAcls)
+		if err := d.ResourceAcls[i].decode(pd, version); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (d *DescribeAclsResponse) key() int16 {
+	return 29
+}
+
+func (d *DescribeAclsResponse) version() int16 {
+	return 0
+}
+
+func (d *DescribeAclsResponse) requiredVersion() KafkaVersion {
+	return V0_11_0_0
+}

vendor/github.com/Shopify/sarama/acl_describe_response_test.go

+package sarama
+
+import (
+	"testing"
+	"time"
+)
+
+var aclDescribeResponseError = []byte{
+	0, 0, 0, 100,
+	0, 8, // error
+	0, 5, 'e', 'r', 'r', 'o', 'r',
+	0, 0, 0, 1, // 1 resource
+	2, // cluster type
+	0, 5, 't', 'o', 'p', 'i', 'c',
+	0, 0, 0, 1, // 1 acl
+	0, 9, 'p', 'r', 'i', 'n', 'c', 'i', 'p', 'a', 'l',
+	0, 4, 'h', 'o', 's', 't',
+	4, // write
+	3, // allow
+}
+
+func TestAclDescribeResponse(t *testing.T) {
+	errmsg := "error"
+	resp := &DescribeAclsResponse{
+		ThrottleTime: 100 * time.Millisecond,
+		Err:          ErrBrokerNotAvailable,
+		ErrMsg:       &errmsg,
+		ResourceAcls: []*ResourceAcls{{
+			Resource: Resource{
+				ResourceName: "topic",
+				ResourceType: AclResourceTopic,
+			},
+			Acls: []*Acl{
+				{
+					Principal:      "principal",
+					Host:           "host",
+					Operation:      AclOperationWrite,
+					PermissionType: AclPermissionAllow,
+				},
+			},
+		}},
+	}
+
+	testResponse(t, "describe", resp, aclDescribeResponseError)
+}

vendor/github.com/Shopify/sarama/acl_filter.go

+package sarama
+
+type AclFilter struct {
+	ResourceType   AclResourceType
+	ResourceName   *string
+	Principal      *string
+	Host           *string
+	Operation      AclOperation
+	PermissionType AclPermissionType
+}
+
+func (a *AclFilter) encode(pe packetEncoder) error {
+	pe.putInt8(int8(a.ResourceType))
+	if err := pe.putNullableString(a.ResourceName); err != nil {
+		return err
+	}
+	if err := pe.putNullableString(a.Principal); err != nil {
+		return err
+	}
+	if err := pe.putNullableString(a.Host); err != nil {
+		return err
+	}
+	pe.putInt8(int8(a.Operation))
+	pe.putInt8(int8(a.PermissionType))
+
+	return nil
+}
+
+func (a *AclFilter) decode(pd packetDecoder, version int16) (err error) {
+	resourceType, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.ResourceType = AclResourceType(resourceType)
+
+	if a.ResourceName, err = pd.getNullableString(); err != nil {
+		return err
+	}
+
+	if a.Principal, err = pd.getNullableString(); err != nil {
+		return err
+	}
+
+	if a.Host, err = pd.getNullableString(); err != nil {
+		return err
+	}
+
+	operation, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.Operation = AclOperation(operation)
+
+	permissionType, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.PermissionType = AclPermissionType(permissionType)
+
+	return nil
+}

vendor/github.com/Shopify/sarama/acl_types.go

+package sarama
+
+type AclOperation int
+
+// ref: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/acl/AclOperation.java
+const (
+	AclOperationUnknown         AclOperation = 0
+	AclOperationAny             AclOperation = 1
+	AclOperationAll             AclOperation = 2
+	AclOperationRead            AclOperation = 3
+	AclOperationWrite           AclOperation = 4
+	AclOperationCreate          AclOperation = 5
+	AclOperationDelete          AclOperation = 6
+	AclOperationAlter           AclOperation = 7
+	AclOperationDescribe        AclOperation = 8
+	AclOperationClusterAction   AclOperation = 9
+	AclOperationDescribeConfigs AclOperation = 10
+	AclOperationAlterConfigs    AclOperation = 11
+	AclOperationIdempotentWrite AclOperation = 12
+)
+
+type AclPermissionType int
+
+// ref: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/acl/AclPermissionType.java
+const (
+	AclPermissionUnknown AclPermissionType = 0
+	AclPermissionAny     AclPermissionType = 1
+	AclPermissionDeny    AclPermissionType = 2
+	AclPermissionAllow   AclPermissionType = 3
+)
+
+type AclResourceType int
+
+// ref: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/resource/ResourceType.java
+const (
+	AclResourceUnknown         AclResourceType = 0
+	AclResourceAny             AclResourceType = 1
+	AclResourceTopic           AclResourceType = 2
+	AclResourceGroup           AclResourceType = 3
+	AclResourceCluster         AclResourceType = 4
+	AclResourceTransactionalID AclResourceType = 5
+)

vendor/github.com/Shopify/sarama/api_versions_request.go

+package sarama
+
+type ApiVersionsRequest struct {
+}
+
+func (r *ApiVersionsRequest) encode(pe packetEncoder) error {
+	return nil
+}
+
+func (r *ApiVersionsRequest) decode(pd packetDecoder, version int16) (err error) {
+	return nil
+}
+
+func (r *ApiVersionsRequest) key() int16 {
+	return 18
+}
+
+func (r *ApiVersionsRequest) version() int16 {
+	return 0
+}
+
+func (r *ApiVersionsRequest) requiredVersion() KafkaVersion {
+	return V0_10_0_0
+}

vendor/github.com/Shopify/sarama/api_versions_request_test.go

+package sarama
+
+import "testing"
+
+var (
+	apiVersionRequest = []byte{}
+)
+
+func TestApiVersionsRequest(t *testing.T) {
+	var request *ApiVersionsRequest
+
+	request = new(ApiVersionsRequest)
+	testRequest(t, "basic", request, apiVersionRequest)
+}

vendor/github.com/Shopify/sarama/api_versions_response.go

+package sarama
+
+type ApiVersionsResponseBlock struct {
+	ApiKey     int16
+	MinVersion int16
+	MaxVersion int16
+}
+
+func (b *ApiVersionsResponseBlock) encode(pe packetEncoder) error {
+	pe.putInt16(b.ApiKey)
+	pe.putInt16(b.MinVersion)
+	pe.putInt16(b.MaxVersion)
+	return nil
+}
+
+func (b *ApiVersionsResponseBlock) decode(pd packetDecoder) error {
+	var err error
+
+	if b.ApiKey, err = pd.getInt16(); err != nil {
+		return err
+	}
+
+	if b.MinVersion, err = pd.getInt16(); err != nil {
+		return err
+	}
+
+	if b.MaxVersion, err = pd.getInt16(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+type ApiVersionsResponse struct {
+	Err         KError
+	ApiVersions []*ApiVersionsResponseBlock
+}
+
+func (r *ApiVersionsResponse) encode(pe packetEncoder) error {
+	pe.putInt16(int16(r.Err))
+	if err := pe.putArrayLength(len(r.ApiVersions)); err != nil {
+		return err
+	}
+	for _, apiVersion := range r.ApiVersions {
+		if err := apiVersion.encode(pe); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *ApiVersionsResponse) decode(pd packetDecoder, version int16) error {
+	kerr, err := pd.getInt16()
+	if err != nil {
+		return err
+	}
+
+	r.Err = KError(kerr)
+
+	numBlocks, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	r.ApiVersions = make([]*ApiVersionsResponseBlock, numBlocks)
+	for i := 0; i < numBlocks; i++ {
+		block := new(ApiVersionsResponseBlock)
+		if err := block.decode(pd); err != nil {
+			return err
+		}
+		r.ApiVersions[i] = block
+	}
+
+	return nil
+}
+
+func (r *ApiVersionsResponse) key() int16 {
+	return 18
+}
+
+func (r *ApiVersionsResponse) version() int16 {
+	return 0
+}
+
+func (r *ApiVersionsResponse) requiredVersion() KafkaVersion {
+	return V0_10_0_0
+}

vendor/github.com/Shopify/sarama/api_versions_response_test.go

+package sarama
+
+import "testing"
+
+var (
+	apiVersionResponse = []byte{
+		0x00, 0x00,
+		0x00, 0x00, 0x00, 0x01,
+		0x00, 0x03,
+		0x00, 0x02,
+		0x00, 0x01,
+	}
+)
+
+func TestApiVersionsResponse(t *testing.T) {
+	var response *ApiVersionsResponse
+
+	response = new(ApiVersionsResponse)
+	testVersionDecodable(t, "no error", response, apiVersionResponse, 0)
+	if response.Err != ErrNoError {
+		t.Error("Decoding error failed: no error expected but found", response.Err)
+	}
+	if response.ApiVersions[0].ApiKey != 0x03 {
+		t.Error("Decoding error: expected 0x03 but got", response.ApiVersions[0].ApiKey)
+	}
+	if response.ApiVersions[0].MinVersion != 0x02 {
+		t.Error("Decoding error: expected 0x02 but got", response.ApiVersions[0].MinVersion)
+	}
+	if response.ApiVersions[0].MaxVersion != 0x01 {
+		t.Error("Decoding error: expected 0x01 but got", response.ApiVersions[0].MaxVersion)
+	}
+}

vendor/github.com/Shopify/sarama/broker_test.go

+package sarama
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+func ExampleBroker() {
+	broker := NewBroker("localhost:9092")
+	err := broker.Open(nil)
+	if err != nil {
+		panic(err)
+	}
+
+	request := MetadataRequest{Topics: []string{"myTopic"}}
+	response, err := broker.GetMetadata(&request)
+	if err != nil {
+		_ = broker.Close()
+		panic(err)
+	}
+
+	fmt.Println("There are", len(response.Topics), "topics active in the cluster.")
+
+	if err = broker.Close(); err != nil {
+		panic(err)
+	}
+}
+
+type mockEncoder struct {
+	bytes []byte
+}
+
+func (m mockEncoder) encode(pe packetEncoder) error {
+	return pe.putRawBytes(m.bytes)
+}
+
+type brokerMetrics struct {
+	bytesRead    int
+	bytesWritten int
+}
+
+func TestBrokerAccessors(t *testing.T) {
+	broker := NewBroker("abc:123")
+
+	if broker.ID() != -1 {
+		t.Error("New broker didn't have an ID of -1.")
+	}
+
+	if broker.Addr() != "abc:123" {
+		t.Error("New broker didn't have the correct address")
+	}
+
+	broker.id = 34
+	if broker.ID() != 34 {
+		t.Error("Manually setting broker ID did not take effect.")
+	}
+}
+
+func TestSimpleBrokerCommunication(t *testing.T) {
+	for _, tt := range brokerTestTable {
+		Logger.Printf("Testing broker communication for %s", tt.name)
+		mb := NewMockBroker(t, 0)
+		mb.Returns(&mockEncoder{tt.response})
+		pendingNotify := make(chan brokerMetrics)
+		// Register a callback to be notified about successful requests
+		mb.SetNotifier(func(bytesRead, bytesWritten int) {
+			pendingNotify <- brokerMetrics{bytesRead, bytesWritten}
+		})
+		broker := NewBroker(mb.Addr())
+		// Set the broker id in order to validate local broker metrics
+		broker.id = 0
+		conf := NewConfig()
+		conf.Version = V0_10_0_0
+		err := broker.Open(conf)
+		if err != nil {
+			t.Fatal(err)
+		}
+		tt.runner(t, broker)
+		// Wait up to 500 ms for the remote broker to process the request and
+		// notify us about the metrics
+		timeout := 500 * time.Millisecond
+		select {
+		case mockBrokerMetrics := <-pendingNotify:
+			validateBrokerMetrics(t, broker, mockBrokerMetrics)
+		case <-time.After(timeout):
+			t.Errorf("No request received for: %s after waiting for %v", tt.name, timeout)
+		}
+		mb.Close()
+		err = broker.Close()
+		if err != nil {
+			t.Error(err)
+		}
+	}
+
+}
+
+// We're not testing encoding/decoding here, so most of the requests/responses will be empty for simplicity's sake
+var brokerTestTable = []struct {
+	name     string
+	response []byte
+	runner   func(*testing.T, *Broker)
+}{
+	{"MetadataRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := MetadataRequest{}
+			response, err := broker.GetMetadata(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("Metadata request got no response!")
+			}
+		}},
+
+	{"ConsumerMetadataRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 't', 0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := ConsumerMetadataRequest{}
+			response, err := broker.GetConsumerMetadata(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("Consumer Metadata request got no response!")
+			}
+		}},
+
+	{"ProduceRequest (NoResponse)",
+		[]byte{},
+		func(t *testing.T, broker *Broker) {
+			request := ProduceRequest{}
+			request.RequiredAcks = NoResponse
+			response, err := broker.Produce(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response != nil {
+				t.Error("Produce request with NoResponse got a response!")
+			}
+		}},
+
+	{"ProduceRequest (WaitForLocal)",
+		[]byte{0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := ProduceRequest{}
+			request.RequiredAcks = WaitForLocal
+			response, err := broker.Produce(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("Produce request without NoResponse got no response!")
+			}
+		}},
+
+	{"FetchRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := FetchRequest{}
+			response, err := broker.Fetch(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("Fetch request got no response!")
+			}
+		}},
+
+	{"OffsetFetchRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := OffsetFetchRequest{}
+			response, err := broker.FetchOffset(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("OffsetFetch request got no response!")
+			}
+		}},
+
+	{"OffsetCommitRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := OffsetCommitRequest{}
+			response, err := broker.CommitOffset(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("OffsetCommit request got no response!")
+			}
+		}},
+
+	{"OffsetRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := OffsetRequest{}
+			response, err := broker.GetAvailableOffsets(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("Offset request got no response!")
+			}
+		}},
+
+	{"JoinGroupRequest",
+		[]byte{0x00, 0x17, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := JoinGroupRequest{}
+			response, err := broker.JoinGroup(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("JoinGroup request got no response!")
+			}
+		}},
+
+	{"SyncGroupRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := SyncGroupRequest{}
+			response, err := broker.SyncGroup(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("SyncGroup request got no response!")
+			}
+		}},
+
+	{"LeaveGroupRequest",
+		[]byte{0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := LeaveGroupRequest{}
+			response, err := broker.LeaveGroup(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("LeaveGroup request got no response!")
+			}
+		}},
+
+	{"HeartbeatRequest",
+		[]byte{0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := HeartbeatRequest{}
+			response, err := broker.Heartbeat(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("Heartbeat request got no response!")
+			}
+		}},
+
+	{"ListGroupsRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := ListGroupsRequest{}
+			response, err := broker.ListGroups(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("ListGroups request got no response!")
+			}
+		}},
+
+	{"DescribeGroupsRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := DescribeGroupsRequest{}
+			response, err := broker.DescribeGroups(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("DescribeGroups request got no response!")
+			}
+		}},
+
+	{"ApiVersionsRequest",
+		[]byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00},
+		func(t *testing.T, broker *Broker) {
+			request := ApiVersionsRequest{}
+			response, err := broker.ApiVersions(&request)
+			if err != nil {
+				t.Error(err)
+			}
+			if response == nil {
+				t.Error("ApiVersions request got no response!")
+			}
+		}},
+}
+
+func validateBrokerMetrics(t *testing.T, broker *Broker, mockBrokerMetrics brokerMetrics) {
+	metricValidators := newMetricValidators()
+	mockBrokerBytesRead := mockBrokerMetrics.bytesRead
+	mockBrokerBytesWritten := mockBrokerMetrics.bytesWritten
+
+	// Check that the number of bytes sent corresponds to what the mock broker received
+	metricValidators.registerForAllBrokers(broker, countMeterValidator("incoming-byte-rate", mockBrokerBytesWritten))
+	if mockBrokerBytesWritten == 0 {
+		// This a ProduceRequest with NoResponse
+		metricValidators.registerForAllBrokers(broker, countMeterValidator("response-rate", 0))
+		metricValidators.registerForAllBrokers(broker, countHistogramValidator("response-size", 0))
+		metricValidators.registerForAllBrokers(broker, minMaxHistogramValidator("response-size", 0, 0))
+	} else {
+		metricValidators.registerForAllBrokers(broker, countMeterValidator("response-rate", 1))
+		metricValidators.registerForAllBrokers(broker, countHistogramValidator("response-size", 1))
+		metricValidators.registerForAllBrokers(broker, minMaxHistogramValidator("response-size", mockBrokerBytesWritten, mockBrokerBytesWritten))
+	}
+
+	// Check that the number of bytes received corresponds to what the mock broker sent
+	metricValidators.registerForAllBrokers(broker, countMeterValidator("outgoing-byte-rate", mockBrokerBytesRead))
+	metricValidators.registerForAllBrokers(broker, countMeterValidator("request-rate", 1))
+	metricValidators.registerForAllBrokers(broker, countHistogramValidator("request-size", 1))
+	metricValidators.registerForAllBrokers(broker, minMaxHistogramValidator("request-size", mockBrokerBytesRead, mockBrokerBytesRead))
+
+	// Run the validators
+	metricValidators.run(t, broker.conf.MetricRegistry)
+}

vendor/github.com/Shopify/sarama/client_tls_test.go

+package sarama
+
+import (
+	"math/big"
+	"net"
+	"testing"
+	"time"
+
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+)
+
+func TestTLS(t *testing.T) {
+	cakey, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	clientkey, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	hostkey, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	nvb := time.Now().Add(-1 * time.Hour)
+	nva := time.Now().Add(1 * time.Hour)
+
+	caTemplate := &x509.Certificate{
+		Subject:      pkix.Name{CommonName: "ca"},
+		Issuer:       pkix.Name{CommonName: "ca"},
+		SerialNumber: big.NewInt(0),
+		NotAfter:     nva,
+		NotBefore:    nvb,
+		IsCA:         true,
+		BasicConstraintsValid: true,
+		KeyUsage:              x509.KeyUsageCertSign,
+	}
+	caDer, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, &cakey.PublicKey, cakey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	caFinalCert, err := x509.ParseCertificate(caDer)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	hostDer, err := x509.CreateCertificate(rand.Reader, &x509.Certificate{
+		Subject:      pkix.Name{CommonName: "host"},
+		Issuer:       pkix.Name{CommonName: "ca"},
+		IPAddresses:  []net.IP{net.IPv4(127, 0, 0, 1)},
+		SerialNumber: big.NewInt(0),
+		NotAfter:     nva,
+		NotBefore:    nvb,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+	}, caFinalCert, &hostkey.PublicKey, cakey)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	clientDer, err := x509.CreateCertificate(rand.Reader, &x509.Certificate{
+		Subject:      pkix.Name{CommonName: "client"},
+		Issuer:       pkix.Name{CommonName: "ca"},
+		SerialNumber: big.NewInt(0),
+		NotAfter:     nva,
+		NotBefore:    nvb,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
+	}, caFinalCert, &clientkey.PublicKey, cakey)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pool := x509.NewCertPool()
+	pool.AddCert(caFinalCert)
+
+	systemCerts, err := x509.SystemCertPool()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Keep server the same - it's the client that we're testing
+	serverTLSConfig := &tls.Config{
+		Certificates: []tls.Certificate{tls.Certificate{
+			Certificate: [][]byte{hostDer},
+			PrivateKey:  hostkey,
+		}},
+		ClientAuth: tls.RequireAndVerifyClientCert,
+		ClientCAs:  pool,
+	}
+
+	for _, tc := range []struct {
+		Succeed        bool
+		Server, Client *tls.Config
+	}{
+		{ // Verify client fails if wrong CA cert pool is specified
+			Succeed: false,
+			Server:  serverTLSConfig,
+			Client: &tls.Config{
+				RootCAs: systemCerts,
+				Certificates: []tls.Certificate{tls.Certificate{
+					Certificate: [][]byte{clientDer},
+					PrivateKey:  clientkey,
+				}},
+			},
+		},
+		{ // Verify client fails if wrong key is specified
+			Succeed: false,
+			Server:  serverTLSConfig,
+			Client: &tls.Config{
+				RootCAs: pool,
+				Certificates: []tls.Certificate{tls.Certificate{
+					Certificate: [][]byte{clientDer},
+					PrivateKey:  hostkey,
+				}},
+			},
+		},
+		{ // Verify client fails if wrong cert is specified
+			Succeed: false,
+			Server:  serverTLSConfig,
+			Client: &tls.Config{
+				RootCAs: pool,
+				Certificates: []tls.Certificate{tls.Certificate{
+					Certificate: [][]byte{hostDer},
+					PrivateKey:  clientkey,
+				}},
+			},
+		},
+		{ // Verify client fails if no CAs are specified
+			Succeed: false,
+			Server:  serverTLSConfig,
+			Client: &tls.Config{
+				Certificates: []tls.Certificate{tls.Certificate{
+					Certificate: [][]byte{clientDer},
+					PrivateKey:  clientkey,
+				}},
+			},
+		},
+		{ // Verify client fails if no keys are specified
+			Succeed: false,
+			Server:  serverTLSConfig,
+			Client: &tls.Config{
+				RootCAs: pool,
+			},
+		},
+		{ // Finally, verify it all works happily with client and server cert in place
+			Succeed: true,
+			Server:  serverTLSConfig,
+			Client: &tls.Config{
+				RootCAs: pool,
+				Certificates: []tls.Certificate{tls.Certificate{
+					Certificate: [][]byte{clientDer},
+					PrivateKey:  clientkey,
+				}},
+			},
+		},
+	} {
+		doListenerTLSTest(t, tc.Succeed, tc.Server, tc.Client)
+	}
+}
+
+func doListenerTLSTest(t *testing.T, expectSuccess bool, serverConfig, clientConfig *tls.Config) {
+	serverConfig.BuildNameToCertificate()
+	clientConfig.BuildNameToCertificate()
+
+	seedListener, err := tls.Listen("tcp", "127.0.0.1:0", serverConfig)
+	if err != nil {
+		t.Fatal("cannot open listener", err)
+	}
+
+	var childT *testing.T
+	if expectSuccess {
+		childT = t
+	} else {
+		childT = &testing.T{} // we want to swallow errors
+	}
+
+	seedBroker := NewMockBrokerListener(childT, 1, seedListener)
+	defer seedBroker.Close()
+
+	seedBroker.Returns(new(MetadataResponse))
+
+	config := NewConfig()
+	config.Net.TLS.Enable = true
+	config.Net.TLS.Config = clientConfig
+
+	client, err := NewClient([]string{seedBroker.Addr()}, config)
+	if err == nil {
+		safeClose(t, client)
+	}
+
+	if expectSuccess {
+		if err != nil {
+			t.Fatal(err)
+		}
+	} else {
+		if err == nil {
+			t.Fatal("expected failure")
+		}
+	}
+}

vendor/github.com/Shopify/sarama/config_test.go

+package sarama
+
+import (
+	"os"
+	"testing"
+
+	"github.com/rcrowley/go-metrics"
+)
+
+func TestDefaultConfigValidates(t *testing.T) {
+	config := NewConfig()
+	if err := config.Validate(); err != nil {
+		t.Error(err)
+	}
+	if config.MetricRegistry == nil {
+		t.Error("Expected non nil metrics.MetricRegistry, got nil")
+	}
+}
+
+func TestInvalidClientIDConfigValidates(t *testing.T) {
+	config := NewConfig()
+	config.ClientID = "foo:bar"
+	if err := config.Validate(); string(err.(ConfigurationError)) != "ClientID is invalid" {
+		t.Error("Expected invalid ClientID, got ", err)
+	}
+}
+
+func TestEmptyClientIDConfigValidates(t *testing.T) {
+	config := NewConfig()
+	config.ClientID = ""
+	if err := config.Validate(); string(err.(ConfigurationError)) != "ClientID is invalid" {
+		t.Error("Expected invalid ClientID, got ", err)
+	}
+}
+
+func TestNetConfigValidates(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  func(*Config) // resorting to using a function as a param because of internal composite structs
+		err  string
+	}{
+		{
+			"OpenRequests",
+			func(cfg *Config) {
+				cfg.Net.MaxOpenRequests = 0
+			},
+			"Net.MaxOpenRequests must be > 0"},
+		{"DialTimeout",
+			func(cfg *Config) {
+				cfg.Net.DialTimeout = 0
+			},
+			"Net.DialTimeout must be > 0"},
+		{"ReadTimeout",
+			func(cfg *Config) {
+				cfg.Net.ReadTimeout = 0
+			},
+			"Net.ReadTimeout must be > 0"},
+		{"WriteTimeout",
+			func(cfg *Config) {
+				cfg.Net.WriteTimeout = 0
+			},
+			"Net.WriteTimeout must be > 0"},
+		{"KeepAlive",
+			func(cfg *Config) {
+				cfg.Net.KeepAlive = -1
+			},
+			"Net.KeepAlive must be >= 0"},
+		{"SASL.User",
+			func(cfg *Config) {
+				cfg.Net.SASL.Enable = true
+				cfg.Net.SASL.User = ""
+			},
+			"Net.SASL.User must not be empty when SASL is enabled"},
+		{"SASL.Password",
+			func(cfg *Config) {
+				cfg.Net.SASL.Enable = true
+				cfg.Net.SASL.User = "user"
+				cfg.Net.SASL.Password = ""
+			},
+			"Net.SASL.Password must not be empty when SASL is enabled"},
+	}
+
+	for i, test := range tests {
+		c := NewConfig()
+		test.cfg(c)
+		if err := c.Validate(); string(err.(ConfigurationError)) != test.err {
+			t.Errorf("[%d]:[%s] Expected %s, Got %s\n", i, test.name, test.err, err)
+		}
+	}
+}
+
+func TestMetadataConfigValidates(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  func(*Config) // resorting to using a function as a param because of internal composite structs
+		err  string
+	}{
+		{
+			"Retry.Max",
+			func(cfg *Config) {
+				cfg.Metadata.Retry.Max = -1
+			},
+			"Metadata.Retry.Max must be >= 0"},
+		{"Retry.Backoff",
+			func(cfg *Config) {
+				cfg.Metadata.Retry.Backoff = -1
+			},
+			"Metadata.Retry.Backoff must be >= 0"},
+		{"RefreshFrequency",
+			func(cfg *Config) {
+				cfg.Metadata.RefreshFrequency = -1
+			},
+			"Metadata.RefreshFrequency must be >= 0"},
+	}
+
+	for i, test := range tests {
+		c := NewConfig()
+		test.cfg(c)
+		if err := c.Validate(); string(err.(ConfigurationError)) != test.err {
+			t.Errorf("[%d]:[%s] Expected %s, Got %s\n", i, test.name, test.err, err)
+		}
+	}
+}
+
+func TestProducerConfigValidates(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  func(*Config) // resorting to using a function as a param because of internal composite structs
+		err  string
+	}{
+		{
+			"MaxMessageBytes",
+			func(cfg *Config) {
+				cfg.Producer.MaxMessageBytes = 0
+			},
+			"Producer.MaxMessageBytes must be > 0"},
+		{"RequiredAcks",
+			func(cfg *Config) {
+				cfg.Producer.RequiredAcks = -2
+			},
+			"Producer.RequiredAcks must be >= -1"},
+		{"Timeout",
+			func(cfg *Config) {
+				cfg.Producer.Timeout = 0
+			},
+			"Producer.Timeout must be > 0"},
+		{"Partitioner",
+			func(cfg *Config) {
+				cfg.Producer.Partitioner = nil
+			},
+			"Producer.Partitioner must not be nil"},
+		{"Flush.Bytes",
+			func(cfg *Config) {
+				cfg.Producer.Flush.Bytes = -1
+			},
+			"Producer.Flush.Bytes must be >= 0"},
+		{"Flush.Messages",
+			func(cfg *Config) {
+				cfg.Producer.Flush.Messages = -1
+			},
+			"Producer.Flush.Messages must be >= 0"},
+		{"Flush.Frequency",
+			func(cfg *Config) {
+				cfg.Producer.Flush.Frequency = -1
+			},
+			"Producer.Flush.Frequency must be >= 0"},
+		{"Flush.MaxMessages",
+			func(cfg *Config) {
+				cfg.Producer.Flush.MaxMessages = -1
+			},
+			"Producer.Flush.MaxMessages must be >= 0"},
+		{"Flush.MaxMessages with Producer.Flush.Messages",
+			func(cfg *Config) {
+				cfg.Producer.Flush.MaxMessages = 1
+				cfg.Producer.Flush.Messages = 2
+			},
+			"Producer.Flush.MaxMessages must be >= Producer.Flush.Messages when set"},
+		{"Flush.Retry.Max",
+			func(cfg *Config) {
+				cfg.Producer.Retry.Max = -1
+			},
+			"Producer.Retry.Max must be >= 0"},
+		{"Flush.Retry.Backoff",
+			func(cfg *Config) {
+				cfg.Producer.Retry.Backoff = -1
+			},
+			"Producer.Retry.Backoff must be >= 0"},
+	}
+
+	for i, test := range tests {
+		c := NewConfig()
+		test.cfg(c)
+		if err := c.Validate(); string(err.(ConfigurationError)) != test.err {
+			t.Errorf("[%d]:[%s] Expected %s, Got %s\n", i, test.name, test.err, err)
+		}
+	}
+}
+
+func TestLZ4ConfigValidation(t *testing.T) {
+	config := NewConfig()
+	config.Producer.Compression = CompressionLZ4
+	if err := config.Validate(); string(err.(ConfigurationError)) != "lz4 compression requires Version >= V0_10_0_0" {
+		t.Error("Expected invalid lz4/kakfa version error, got ", err)
+	}
+	config.Version = V0_10_0_0
+	if err := config.Validate(); err != nil {
+		t.Error("Expected lz4 to work, got ", err)
+	}
+}
+
+// This example shows how to integrate with an existing registry as well as publishing metrics
+// on the standard output
+func ExampleConfig_metrics() {
+	// Our application registry
+	appMetricRegistry := metrics.NewRegistry()
+	appGauge := metrics.GetOrRegisterGauge("m1", appMetricRegistry)
+	appGauge.Update(1)
+
+	config := NewConfig()
+	// Use a prefix registry instead of the default local one
+	config.MetricRegistry = metrics.NewPrefixedChildRegistry(appMetricRegistry, "sarama.")
+
+	// Simulate a metric created by sarama without starting a broker
+	saramaGauge := metrics.GetOrRegisterGauge("m2", config.MetricRegistry)
+	saramaGauge.Update(2)
+
+	metrics.WriteOnce(appMetricRegistry, os.Stdout)
+	// Output:
+	// gauge m1
+	//   value:               1
+	// gauge sarama.m2
+	//   value:               2
+}

vendor/github.com/Shopify/sarama/consumer_metadata_request.go

+package sarama
+
+type ConsumerMetadataRequest struct {
+	ConsumerGroup string
+}
+
+func (r *ConsumerMetadataRequest) encode(pe packetEncoder) error {
+	return pe.putString(r.ConsumerGroup)
+}
+
+func (r *ConsumerMetadataRequest) decode(pd packetDecoder, version int16) (err error) {
+	r.ConsumerGroup, err = pd.getString()
+	return err
+}
+
+func (r *ConsumerMetadataRequest) key() int16 {
+	return 10
+}
+
+func (r *ConsumerMetadataRequest) version() int16 {
+	return 0
+}
+
+func (r *ConsumerMetadataRequest) requiredVersion() KafkaVersion {
+	return V0_8_2_0
+}

vendor/github.com/Shopify/sarama/dev.yml

+name: sarama
+
+up:
+  - go:
+      version: '1.9'
+
+commands:
+  test:
+    run: make test
+    desc: 'run unit tests'