chore(lint): re-enable a small amount of go-critic

...and fix warnings

Author: dnwe

.golangci.yml

@@ -23,13 +23,18 @@ linters-settings:
   gocritic:
     enabled-tags:
       - diagnostic
-      - experimental
-      - opinionated
-      - performance
-      - style
+      # - experimental
+      # - opinionated
+      # - performance
+      # - style
     disabled-checks:
-      - wrapperFunc
+      - assignOp
+      - appendAssign
+      - commentedOutCode
       - ifElseChain
+      - singleCaseSwitch
+      - sloppyReassign
+      - wrapperFunc
   funlen:
     lines: 300
     statements: 300
@@ -48,7 +53,7 @@ linters:
     - funlen
     - gochecknoinits
     # - goconst
-    # - gocritic
+    - gocritic
     - gocyclo
     - gofmt
     - goimports

admin_test.go

@@ -161,7 +161,7 @@ func TestClusterAdminListTopics(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if len(entries) <= 0 {
+	if len(entries) == 0 {
 		t.Fatal(errors.New("no resource present"))
 	}
 
@@ -447,10 +447,10 @@ func TestClusterAdminListPartitionReassignments(t *testing.T) {
 	partitionStatus, ok := response["my_topic"]
 	if !ok {
 		t.Fatalf("topic missing in response")
-	} else {
-		if len(partitionStatus) != 2 {
-			t.Fatalf("partition missing in response")
-		}
+	}
+
+	if len(partitionStatus) != 2 {
+		t.Fatalf("partition missing in response")
 	}
 
 	err = admin.Close()
@@ -711,47 +711,49 @@ func TestClusterAdminDescribeConfig(t *testing.T) {
 		{V2_0_0_0, 2, true},
 	}
 	for _, tt := range tests {
-		config := NewTestConfig()
-		config.Version = tt.saramaVersion
-		admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer func() {
-			_ = admin.Close()
-		}()
-
-		resource := ConfigResource{
-			Name:        "r1",
-			Type:        TopicResource,
-			ConfigNames: []string{"my_topic"},
-		}
+		t.Run(tt.saramaVersion.String(), func(t *testing.T) {
+			config := NewTestConfig()
+			config.Version = tt.saramaVersion
+			admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer func() {
+				_ = admin.Close()
+			}()
+
+			resource := ConfigResource{
+				Name:        "r1",
+				Type:        TopicResource,
+				ConfigNames: []string{"my_topic"},
+			}
 
-		entries, err := admin.DescribeConfig(resource)
-		if err != nil {
-			t.Fatal(err)
-		}
+			entries, err := admin.DescribeConfig(resource)
+			if err != nil {
+				t.Fatal(err)
+			}
 
-		history := seedBroker.History()
-		describeReq, ok := history[len(history)-1].Request.(*DescribeConfigsRequest)
-		if !ok {
-			t.Fatal("failed to find DescribeConfigsRequest in mockBroker history")
-		}
+			history := seedBroker.History()
+			describeReq, ok := history[len(history)-1].Request.(*DescribeConfigsRequest)
+			if !ok {
+				t.Fatal("failed to find DescribeConfigsRequest in mockBroker history")
+			}
 
-		if describeReq.Version != tt.requestVersion {
-			t.Fatalf(
-				"requestVersion %v did not match expected %v",
-				describeReq.Version, tt.requestVersion)
-		}
+			if describeReq.Version != tt.requestVersion {
+				t.Fatalf(
+					"requestVersion %v did not match expected %v",
+					describeReq.Version, tt.requestVersion)
+			}
 
-		if len(entries) <= 0 {
-			t.Fatal(errors.New("no resource present"))
-		}
-		if tt.includeSynonyms {
-			if len(entries[0].Synonyms) == 0 {
-				t.Fatal("expected synonyms to have been included")
+			if len(entries) == 0 {
+				t.Fatal(errors.New("no resource present"))
 			}
-		}
+			if tt.includeSynonyms {
+				if len(entries[0].Synonyms) == 0 {
+					t.Fatal("expected synonyms to have been included")
+				}
+			}
+		})
 	}
 }
 
@@ -829,7 +831,7 @@ func TestClusterAdminDescribeBrokerConfig(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		if len(entries) <= 0 {
+		if len(entries) == 0 {
 			t.Fatal(errors.New("no resource present"))
 		}
 	}
@@ -1245,7 +1247,7 @@ func TestClusterAdminListAcls(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(rAcls) <= 0 {
+	if len(rAcls) == 0 {
 		t.Fatal("no acls present")
 	}
 

async_producer_test.go

@@ -684,7 +684,7 @@ func TestAsyncProducerMultipleRetriesWithBackoffFunc(t *testing.T) {
 
 // https://github.com/Shopify/sarama/issues/2129
 func TestAsyncProducerMultipleRetriesWithConcurrentRequests(t *testing.T) {
-	//Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
+	// Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
 	seedBroker := NewMockBroker(t, 1)
 	leader := NewMockBroker(t, 2)
 

broker.go

@@ -1407,7 +1407,7 @@ func (b *Broker) sendAndReceiveSASLSCRAMv0() error {
 		// Will be decremented in updateIncomingCommunicationMetrics (except error)
 		b.addRequestInFlightMetrics(1)
 		length := len(msg)
-		authBytes := make([]byte, length+4) //4 byte length header + auth data
+		authBytes := make([]byte, length+4) // 4 byte length header + auth data
 		binary.BigEndian.PutUint32(authBytes, uint32(length))
 		copy(authBytes[4:], []byte(msg))
 		_, err := b.write(authBytes)

client.go

@@ -135,7 +135,7 @@ type client struct {
 
 	lock sync.RWMutex // protects access to the maps that hold cluster state.
 
-	updateMetaDataMs int64 //store update metadata time
+	updateMetaDataMs int64 // store update metadata time
 }
 
 // NewClient creates a new Client. It connects to one of the given broker addresses

delete_offsets_response.go

@@ -5,10 +5,10 @@ import (
 )
 
 type DeleteOffsetsResponse struct {
-	//The top-level error code, or 0 if there was no error.
+	// The top-level error code, or 0 if there was no error.
 	ErrorCode    KError
 	ThrottleTime time.Duration
-	//The responses for each partition of the topics.
+	// The responses for each partition of the topics.
 	Errors map[string]map[int32]KError
 }
 

functional_consumer_test.go

@@ -553,12 +553,12 @@ func consumeMsgs(t *testing.T, clientVersions []KafkaVersion, producedMessages [
 		if err != nil {
 			t.Fatal(err)
 		}
-		defer safeClose(t, c)
+		defer safeClose(t, c) //nolint: gocritic // the close intentionally happens outside the loop
 		pc, err := c.ConsumePartition("test.1", 0, producedMessages[0].Offset)
 		if err != nil {
 			t.Fatal(err)
 		}
-		defer safeClose(t, pc)
+		defer safeClose(t, pc) //nolint: gocritic // the close intentionally happens outside the loop
 
 		var wg sync.WaitGroup
 		wg.Add(1)

mocks/async_producer.go

@@ -85,10 +85,8 @@ func NewAsyncProducer(t ErrorReporter, config *sarama.Config) *AsyncProducer {
 							msg.Offset = mp.lastOffset
 							mp.successes <- msg
 						}
-					} else {
-						if config.Producer.Return.Errors {
-							mp.errors <- &sarama.ProducerError{Err: expectation.Result, Msg: msg}
-						}
+					} else if config.Producer.Return.Errors {
+						mp.errors <- &sarama.ProducerError{Err: expectation.Result, Msg: msg}
 					}
 				}
 			}

offset_commit_response_test.go

@@ -28,7 +28,7 @@ var (
 )
 
 func TestEmptyOffsetCommitResponse(t *testing.T) {
-	//groupInstanceId := "gid"
+	// groupInstanceId := "gid"
 	tests := []struct {
 		CaseName     string
 		Version      int16

partitioner_test.go

@@ -247,7 +247,8 @@ func ExamplePartitioner_random() {
 
 	producer, err := NewSyncProducer([]string{"localhost:9092"}, config)
 	if err != nil {
-		log.Fatal(err)
+		log.Println(err)
+		return
 	}
 	defer func() {
 		if err := producer.Close(); err != nil {
@@ -258,7 +259,8 @@ func ExamplePartitioner_random() {
 	msg := &ProducerMessage{Topic: "test", Key: StringEncoder("key is set"), Value: StringEncoder("test")}
 	partition, offset, err := producer.SendMessage(msg)
 	if err != nil {
-		log.Fatalln("Failed to produce message to kafka cluster.")
+		log.Println("Failed to produce message to kafka cluster.")
+		return
 	}
 
 	log.Printf("Produced message to partition %d with offset %d", partition, offset)
@@ -273,7 +275,8 @@ func ExamplePartitioner_manual() {
 
 	producer, err := NewSyncProducer([]string{"localhost:9092"}, config)
 	if err != nil {
-		log.Fatal(err)
+		log.Println(err)
+		return
 	}
 	defer func() {
 		if err := producer.Close(); err != nil {
@@ -286,11 +289,13 @@ func ExamplePartitioner_manual() {
 
 	partition, offset, err := producer.SendMessage(msg)
 	if err != nil {
-		log.Fatalln("Failed to produce message to kafka cluster.")
+		log.Println("Failed to produce message to kafka cluster.")
+		return
 	}
 
 	if partition != 6 {
-		log.Fatal("Message should have been produced to partition 6!")
+		log.Println("Message should have been produced to partition 6!")
+		return
 	}
 
 	log.Printf("Produced message to partition %d with offset %d", partition, offset)

records.go

@@ -203,7 +203,7 @@ func magicValue(pd packetDecoder) (int8, error) {
 }
 
 func (r *Records) getControlRecord() (ControlRecord, error) {
-	if r.RecordBatch == nil || len(r.RecordBatch.Records) <= 0 {
+	if r.RecordBatch == nil || len(r.RecordBatch.Records) == 0 {
 		return ControlRecord{}, fmt.Errorf("cannot get control record, record batch is empty")
 	}