Merge pull request #1621 from Shopify/diego_golangci-lint-cleanup

enables some golanci lint and fixes the issues

.golangci.yml

@@ -38,32 +38,32 @@ linters:
   disable-all: true
   enable:
     - bodyclose
-    # - deadcode
+    - deadcode
     - depguard
     - dogsled
     # - dupl
-    # - errcheck
+    - errcheck
     - funlen
     # - gocritic
     - gocyclo
     - gofmt
-    # - goimports
+    - goimports
     # - golint
-    # - gosec
+    - gosec
     # - gosimple
     - govet
     # - ineffassign
-    # - interfacer
+    - interfacer
     # - misspell
     # - nakedret
     # - scopelint
     # - staticcheck
-    # - structcheck
+    - structcheck
     # - stylecheck
     - typecheck
-    # - unconvert
-    # - unused
-    # - varcheck
+    - unconvert
+    - unused
+    - varcheck
     - whitespace
     # - goconst
     # - gochecknoinits

acl_delete_response.go

@@ -53,7 +53,7 @@ func (d *DeleteAclsResponse) key() int16 {
 }
 
 func (d *DeleteAclsResponse) version() int16 {
-	return int16(d.Version)
+	return d.Version
 }
 
 func (d *DeleteAclsResponse) requiredVersion() KafkaVersion {

acl_describe_response.go

@@ -74,7 +74,7 @@ func (d *DescribeAclsResponse) key() int16 {
 }
 
 func (d *DescribeAclsResponse) version() int16 {
-	return int16(d.Version)
+	return d.Version
 }
 
 func (d *DescribeAclsResponse) requiredVersion() KafkaVersion {

admin_test.go

@@ -172,7 +172,7 @@ func TestClusterAdminListTopics(t *testing.T) {
 	if found {
 		t.Fatal(errors.New("default topic config entry incorrectly found in response"))
 	}
-	value, _ := topic.ConfigEntries["retention.ms"]
+	value := topic.ConfigEntries["retention.ms"]
 	if value == nil || *value != "5000" {
 		t.Fatal(errors.New("non-default topic config entry not found in response"))
 	}

api_versions_request_test.go

@@ -7,8 +7,6 @@ var (
 )
 
 func TestApiVersionsRequest(t *testing.T) {
-	var request *ApiVersionsRequest
-
-	request = new(ApiVersionsRequest)
+	request := new(ApiVersionsRequest)
 	testRequest(t, "basic", request, apiVersionRequest)
 }

api_versions_response_test.go

@@ -13,9 +13,7 @@ var (
 )
 
 func TestApiVersionsResponse(t *testing.T) {
-	var response *ApiVersionsResponse
-
-	response = new(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)

async_producer.go

@@ -411,7 +411,7 @@ func (tp *topicProducer) partitionMessage(msg *ProducerMessage) error {
 	var partitions []int32
 
 	err := tp.breaker.Run(func() (err error) {
-		var requiresConsistency = false
+		requiresConsistency := false
 		if ep, ok := tp.partitioner.(DynamicConsistencyPartitioner); ok {
 			requiresConsistency = ep.MessageRequiresConsistency(msg)
 		} else {

balance_strategy.go

@@ -267,7 +267,7 @@ func (s *stickyBalanceStrategy) Plan(members map[string]ConsumerGroupMemberMetad
 	plan := make(BalanceStrategyPlan, len(currentAssignment))
 	for memberID, assignments := range currentAssignment {
 		if len(assignments) == 0 {
-			plan[memberID] = make(map[string][]int32, 0)
+			plan[memberID] = make(map[string][]int32)
 		} else {
 			for _, assignment := range assignments {
 				plan.Add(memberID, assignment.Topic, assignment.Partition)
@@ -689,14 +689,6 @@ func sortPartitionsByPotentialConsumerAssignments(partition2AllPotentialConsumer
 	return sortedPartionIDs
 }
 
-func deepCopyPartitions(src []topicPartitionAssignment) []topicPartitionAssignment {
-	dst := make([]topicPartitionAssignment, len(src))
-	for i, partition := range src {
-		dst[i] = partition
-	}
-	return dst
-}
-
 func deepCopyAssignment(assignment map[string][]topicPartitionAssignment) map[string][]topicPartitionAssignment {
 	copy := make(map[string][]topicPartitionAssignment, len(assignment))
 	for memberID, subscriptions := range assignment {
@@ -956,9 +948,7 @@ func (p *partitionMovements) in(cycle []string, cycles [][]string) bool {
 	for i := 0; i < len(cycle)-1; i++ {
 		superCycle[i] = cycle[i]
 	}
-	for _, c := range cycle {
-		superCycle = append(superCycle, c)
-	}
+	superCycle = append(superCycle, cycle...)
 	for _, foundCycle := range cycles {
 		if len(foundCycle) == len(cycle) && indexOfSubList(superCycle, foundCycle) != -1 {
 			return true

balance_strategy_test.go

@@ -2008,7 +2008,7 @@ func Test_stickyBalanceStrategy_Plan_AssignmentData(t *testing.T) {
 	if err != nil {
 		t.Errorf("Error building assignment data: %v", err)
 	}
-	if bytes.Compare(expected, actual) != 0 {
+	if !bytes.Equal(expected, actual) {
 		t.Error("Invalid assignment data returned from AssignmentData")
 	}
 }

broker_test.go

@@ -245,10 +245,6 @@ func TestSASLOAuthBearer(t *testing.T) {
 			if test.expectedBrokerError != err {
 				t.Errorf("[%d]:[%s] Expected %s auth error, got %s\n", i, test.name, test.expectedBrokerError, err)
 			}
-		} else if test.expectedBrokerError != ErrNoError {
-			if test.expectedBrokerError != err {
-				t.Errorf("[%d]:[%s] Expected %s handshake error, got %s\n", i, test.name, test.expectedBrokerError, err)
-			}
 		} else if test.expectClientErr && err == nil {
 			t.Errorf("[%d]:[%s] Expected a client error and got none\n", i, test.name)
 		} else if !test.expectClientErr && err != nil {

compress.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/eapache/go-xerial-snappy"
+	snappy "github.com/eapache/go-xerial-snappy"
 	"github.com/pierrec/lz4"
 )
 

consumer_test.go

@@ -1016,7 +1016,6 @@ func TestConsumerTimestamps(t *testing.T) {
 	now := time.Now().Truncate(time.Millisecond)
 	type testMessage struct {
 		key       Encoder
-		value     Encoder
 		offset    int64
 		timestamp time.Time
 	}
@@ -1027,32 +1026,32 @@ func TestConsumerTimestamps(t *testing.T) {
 		expectedTimestamp []time.Time
 	}{
 		{MinVersion, false, []testMessage{
-			{nil, testMsg, 1, now},
-			{nil, testMsg, 2, now},
+			{testMsg, 1, now},
+			{testMsg, 2, now},
 		}, []time.Time{{}, {}}},
 		{V0_9_0_0, false, []testMessage{
-			{nil, testMsg, 1, now},
-			{nil, testMsg, 2, now},
+			{testMsg, 1, now},
+			{testMsg, 2, now},
 		}, []time.Time{{}, {}}},
 		{V0_10_0_0, false, []testMessage{
-			{nil, testMsg, 1, now},
-			{nil, testMsg, 2, now},
+			{testMsg, 1, now},
+			{testMsg, 2, now},
 		}, []time.Time{{}, {}}},
 		{V0_10_2_1, false, []testMessage{
-			{nil, testMsg, 1, now.Add(time.Second)},
-			{nil, testMsg, 2, now.Add(2 * time.Second)},
+			{testMsg, 1, now.Add(time.Second)},
+			{testMsg, 2, now.Add(2 * time.Second)},
 		}, []time.Time{now.Add(time.Second), now.Add(2 * time.Second)}},
 		{V0_10_2_1, true, []testMessage{
-			{nil, testMsg, 1, now.Add(time.Second)},
-			{nil, testMsg, 2, now.Add(2 * time.Second)},
+			{testMsg, 1, now.Add(time.Second)},
+			{testMsg, 2, now.Add(2 * time.Second)},
 		}, []time.Time{now, now}},
 		{V0_11_0_0, false, []testMessage{
-			{nil, testMsg, 1, now.Add(time.Second)},
-			{nil, testMsg, 2, now.Add(2 * time.Second)},
+			{testMsg, 1, now.Add(time.Second)},
+			{testMsg, 2, now.Add(2 * time.Second)},
 		}, []time.Time{now.Add(time.Second), now.Add(2 * time.Second)}},
 		{V0_11_0_0, true, []testMessage{
-			{nil, testMsg, 1, now.Add(time.Second)},
-			{nil, testMsg, 2, now.Add(2 * time.Second)},
+			{testMsg, 1, now.Add(time.Second)},
+			{testMsg, 2, now.Add(2 * time.Second)},
 		}, []time.Time{now, now}},
 	} {
 		var fr *FetchResponse

decompress.go

@@ -7,7 +7,7 @@ import (
 	"io/ioutil"
 	"sync"
 
-	"github.com/eapache/go-xerial-snappy"
+	snappy "github.com/eapache/go-xerial-snappy"
 	"github.com/pierrec/lz4"
 )
 

describe_configs_request_test.go

@@ -102,9 +102,7 @@ func TestDescribeConfigsRequestv0(t *testing.T) {
 }
 
 func TestDescribeConfigsRequestv1(t *testing.T) {
-	var request *DescribeConfigsRequest
-
-	request = &DescribeConfigsRequest{
+	request := &DescribeConfigsRequest{
 		Version: 1,
 		Resources: []*ConfigResource{
 			{

gssapi_kerberos.go

@@ -242,7 +242,7 @@ func (krbAuth *GSSAPIKerberosAuth) Authorize(broker *Broker) error {
 		}
 		broker.updateOutgoingCommunicationMetrics(bytesWritten)
 		if krbAuth.step == GSS_API_VERIFY {
-			var bytesRead = 0
+			bytesRead := 0
 			receivedBytes, bytesRead, err = krbAuth.readPackage(broker)
 			requestLatency := time.Since(requestTime)
 			broker.updateIncomingCommunicationMetrics(bytesRead, requestLatency)

heartbeat_request_test.go

@@ -11,9 +11,7 @@ var (
 )
 
 func TestHeartbeatRequest(t *testing.T) {
-	var request *HeartbeatRequest
-
-	request = new(HeartbeatRequest)
+	request := new(HeartbeatRequest)
 	request.GroupId = "foo"
 	request.GenerationId = 66051
 	request.MemberId = "baz"

heartbeat_response_test.go

@@ -8,9 +8,7 @@ var (
 )
 
 func TestHeartbeatResponse(t *testing.T) {
-	var response *HeartbeatResponse
-
-	response = new(HeartbeatResponse)
+	response := new(HeartbeatResponse)
 	testVersionDecodable(t, "no error", response, heartbeatResponseNoError, 0)
 	if response.Err != ErrNoError {
 		t.Error("Decoding error failed: no error expected but found", response.Err)

kerberos_client_test.go

@@ -2,9 +2,10 @@ package sarama
 
 import (
 	"errors"
+	"testing"
+
 	krbcfg "gopkg.in/jcmturner/gokrb5.v7/config"
 	"gopkg.in/jcmturner/gokrb5.v7/test/testdata"
-	"testing"
 )
 
 /*

leave_group_request_test.go

@@ -10,9 +10,7 @@ var (
 )
 
 func TestLeaveGroupRequest(t *testing.T) {
-	var request *LeaveGroupRequest
-
-	request = new(LeaveGroupRequest)
+	request := new(LeaveGroupRequest)
 	request.GroupId = "foo"
 	request.MemberId = "bar"
 	testRequest(t, "basic", request, basicLeaveGroupRequest)

metadata_request_test.go

@@ -104,6 +104,9 @@ func TestMetadataRequestV3(t *testing.T) {
 
 	request.Topics = []string{"topic1"}
 	testRequest(t, "one topic", request, metadataRequestOneTopicV3)
+
+	request.Topics = []string{"foo", "bar", "baz"}
+	testRequest(t, "three topics", request, metadataRequestThreeTopicsV3)
 }
 
 func TestMetadataRequestV4(t *testing.T) {
@@ -123,7 +126,7 @@ func TestMetadataRequestV4(t *testing.T) {
 func TestMetadataRequestV5(t *testing.T) {
 	request := new(MetadataRequest)
 	request.Version = 5
-	testRequest(t, "no topics", request, metadataRequestNoTopicsV4)
+	testRequest(t, "no topics", request, metadataRequestNoTopicsV5)
 
 	request.Topics = []string{"topic1"}
 

mockbroker.go

@@ -213,7 +213,7 @@ func (b *MockBroker) isGSSAPI(buffer []byte) bool {
 	return buffer[4] == 0x60 || bytes.Equal(buffer[4:6], []byte{0x05, 0x04})
 }
 
-func (b *MockBroker) handleRequests(conn net.Conn, idx int, wg *sync.WaitGroup) {
+func (b *MockBroker) handleRequests(conn io.ReadWriteCloser, idx int, wg *sync.WaitGroup) {
 	defer wg.Done()
 	defer func() {
 		_ = conn.Close()

mockkerberos.go

@@ -3,6 +3,7 @@ package sarama
 import (
 	"encoding/binary"
 	"encoding/hex"
+
 	"gopkg.in/jcmturner/gokrb5.v7/credentials"
 	"gopkg.in/jcmturner/gokrb5.v7/gssapi"
 	"gopkg.in/jcmturner/gokrb5.v7/iana/keyusage"
@@ -55,7 +56,6 @@ func (h *KafkaGSSAPIHandler) MockKafkaGSSAPI(buffer []byte) []byte {
 }
 
 type MockKerberosClient struct {
-	asReqBytes  string
 	asRepBytes  string
 	ASRep       messages.ASRep
 	credentials *credentials.Credentials

mockresponses.go

@@ -667,7 +667,7 @@ func (mr *MockDeleteTopicsResponse) For(reqBody versionedDecoder) encoder {
 	for _, topic := range req.Topics {
 		res.TopicErrorCodes[topic] = ErrNoError
 	}
-	res.Version = int16(req.Version)
+	res.Version = req.Version
 	return res
 }
 

mocks/async_producer.go

@@ -33,7 +33,7 @@ func NewAsyncProducer(t ErrorReporter, config *sarama.Config) *AsyncProducer {
 	}
 	mp := &AsyncProducer{
 		t:            t,
-		closed:       make(chan struct{}, 0),
+		closed:       make(chan struct{}),
 		expectations: make([]*producerExpectation, 0),
 		input:        make(chan *sarama.ProducerMessage, config.ChannelBufferSize),
 		successes:    make(chan *sarama.ProducerMessage, config.ChannelBufferSize),

mocks/mocks.go

@@ -15,8 +15,6 @@ package mocks
 
 import (
 	"errors"
-
-	"github.com/Shopify/sarama"
 )
 
 // ErrorReporter is a simple interface that includes the testing.T methods we use to report
@@ -41,8 +39,3 @@ type producerExpectation struct {
 	Result        error
 	CheckFunction ValueChecker
 }
-
-type consumerExpectation struct {
-	Err error
-	Msg *sarama.ConsumerMessage
-}

offset_request.go

@@ -6,7 +6,7 @@ type offsetRequestBlock struct {
 }
 
 func (b *offsetRequestBlock) encode(pe packetEncoder, version int16) error {
-	pe.putInt64(int64(b.time))
+	pe.putInt64(b.time)
 	if version == 0 {
 		pe.putInt32(b.maxOffsets)
 	}

real_decoder.go

@@ -7,9 +7,7 @@ import (
 
 var errInvalidArrayLength = PacketDecodingError{"invalid array length"}
 var errInvalidByteSliceLength = PacketDecodingError{"invalid byteslice length"}
-var errInvalidByteSliceLengthType = PacketDecodingError{"invalid byteslice length type"}
 var errInvalidStringLength = PacketDecodingError{"invalid string length"}
-var errInvalidSubsetSize = PacketDecodingError{"invalid subset size"}
 var errVarintOverflow = PacketDecodingError{"varint overflow"}
 var errInvalidBool = PacketDecodingError{"invalid bool"}
 

records.go

@@ -8,7 +8,6 @@ const (
 	defaultRecords
 
 	magicOffset = 16
-	magicLength = 1
 )
 
 // Records implements a union type containing either a RecordBatch or a legacy MessageSet.

sasl_authenticate_request_test.go

@@ -9,9 +9,7 @@ var (
 )
 
 func TestSaslAuthenticateRequest(t *testing.T) {
-	var request *SaslAuthenticateRequest
-
-	request = new(SaslAuthenticateRequest)
+	request := new(SaslAuthenticateRequest)
 	request.SaslAuthBytes = []byte(`foo`)
 	testRequest(t, "basic", request, saslAuthenticateRequest)
 }

sasl_handshake_request_test.go

@@ -9,9 +9,7 @@ var (
 )
 
 func TestSaslHandshakeRequest(t *testing.T) {
-	var request *SaslHandshakeRequest
-
-	request = new(SaslHandshakeRequest)
+	request := new(SaslHandshakeRequest)
 	request.Mechanism = "foo"
 	testRequest(t, "basic", request, baseSaslRequest)
 }

sasl_handshake_response_test.go

@@ -11,9 +11,7 @@ var (
 )
 
 func TestSaslHandshakeResponse(t *testing.T) {
-	var response *SaslHandshakeResponse
-
-	response = new(SaslHandshakeResponse)
+	response := new(SaslHandshakeResponse)
 	testVersionDecodable(t, "no error", response, saslHandshakeResponse, 0)
 	if response.Err != ErrNoError {
 		t.Error("Decoding error failed: no error expected but found", response.Err)

tools/kafka-console-consumer/kafka-console-consumer.go

@@ -9,6 +9,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"syscall"
 
 	"github.com/Shopify/sarama"
 	"github.com/Shopify/sarama/tools/tls"
@@ -85,7 +86,7 @@ func main() {
 
 	go func() {
 		signals := make(chan os.Signal, 1)
-		signal.Notify(signals, os.Kill, os.Interrupt)
+		signal.Notify(signals, syscall.SIGTERM, os.Interrupt)
 		<-signals
 		logger.Println("Initiating shutdown of consumer...")
 		close(closing)

tools/kafka-console-partitionconsumer/kafka-console-partitionconsumer.go

@@ -8,6 +8,7 @@ import (
 	"os/signal"
 	"strconv"
 	"strings"
+	"syscall"
 
 	"github.com/Shopify/sarama"
 )
@@ -70,7 +71,7 @@ func main() {
 
 	go func() {
 		signals := make(chan os.Signal, 1)
-		signal.Notify(signals, os.Kill, os.Interrupt)
+		signal.Notify(signals, syscall.SIGTERM, os.Interrupt)
 		<-signals
 		pc.AsyncClose()
 	}()

tools/kafka-console-producer/kafka-console-producer.go

@@ -8,9 +8,10 @@ import (
 	"os"
 	"strings"
 
+	"github.com/rcrowley/go-metrics"
+
 	"github.com/Shopify/sarama"
 	"github.com/Shopify/sarama/tools/tls"
-	"github.com/rcrowley/go-metrics"
 )
 
 var (

tools/kafka-producer-performance/main.go

@@ -14,9 +14,10 @@ import (
 	gosync "sync"
 	"time"
 
+	metrics "github.com/rcrowley/go-metrics"
+
 	"github.com/Shopify/sarama"
 	"github.com/Shopify/sarama/tools/tls"
-	metrics "github.com/rcrowley/go-metrics"
 )
 
 var (

utils.go

@@ -26,9 +26,7 @@ func (slice int32Slice) Swap(i, j int) {
 
 func dupInt32Slice(input []int32) []int32 {
 	ret := make([]int32, 0, len(input))
-	for _, val := range input {
-		ret = append(ret, val)
-	}
+	ret = append(ret, input...)
 	return ret
 }
 

zstd.go

@@ -1,8 +1,9 @@
 package sarama
 
 import (
-	"github.com/klauspost/compress/zstd"
 	"sync"
+
+	"github.com/klauspost/compress/zstd"
 )
 
 var (