chore: qol improvements (#98)

Author: erayarslan

.golangci.yml

@@ -1,38 +1,48 @@
-linters-settings:
-  lll:
-    line-length: 140
-  funlen:
-    lines: 70
-
+version: "2"
 linters:
-  disable-all: true
+  default: none
   enable:
     - bodyclose
-    - errcheck
     - dupl
+    - errcheck
     - exhaustive
     - funlen
     - goconst
     - gocritic
     - gocyclo
-    - gosimple
     - govet
-    - gosec
     - ineffassign
     - lll
     - misspell
     - nakedret
-    - gofumpt
+    - prealloc
     - staticcheck
-    - stylecheck
-    - typecheck
     - unconvert
     - unparam
     - unused
     - whitespace
-    - prealloc
-
-service:
-  golangci-lint-version: 1.55.2 # use the fixed version to not introduce new linters unexpectedly
-  prepare:
-    - echo "here I can run custom commands, but no preparation needed for this repo"
+  settings:
+    funlen:
+      lines: 70
+    lll:
+      line-length: 140
+  exclusions:
+    generated: lax
+    presets:
+      - comments
+      - common-false-positives
+      - legacy
+      - std-error-handling
+    paths:
+      - third_party$
+      - builtin$
+      - examples$
+formatters:
+  enable:
+    - gofumpt
+  exclusions:
+    generated: lax
+    paths:
+      - third_party$
+      - builtin$
+      - examples$

Makefile

@@ -3,8 +3,8 @@
 default: run
 
 init:
-	go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.55.2
-	go install golang.org/x/tools/go/analysis/passes/fieldalignment/cmd/fieldalignment@v0.15.0
+	go install github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.3.0
+	go install golang.org/x/tools/go/analysis/passes/fieldalignment/cmd/fieldalignment@v0.35.0
 
 clean:
 	rm -rf ./build

connector.go

@@ -197,7 +197,8 @@ func newConnector(cf any, mapper Mapper, sinkResponseHandler dcpElasticsearch.Si
 
 	connector.dcp.SetEventHandler(
 		&DcpEventHandler{
-			bulk: connector.bulk,
+			isFinite: dcpConfig.IsDcpModeFinite(),
+			bulk:     connector.bulk,
 		})
 
 	metricCollector := metric.NewMetricCollector(connector.bulk)

couchbase/event.go

@@ -1,9 +1,10 @@
 package couchbase
 
 import (
-	"github.com/Trendyol/go-dcp/tracing"
 	"time"
 
+	"github.com/Trendyol/go-dcp/tracing"
+
 	"github.com/elastic/go-elasticsearch/v7"
 )
 

dcp_event_handler.go

@@ -5,7 +5,8 @@ import (
 )
 
 type DcpEventHandler struct {
-	bulk *bulk.Bulk
+	bulk     *bulk.Bulk
+	isFinite bool
 }
 
 func (h *DcpEventHandler) BeforeRebalanceStart() {
@@ -28,6 +29,9 @@ func (h *DcpEventHandler) AfterStreamStart() {
 }
 
 func (h *DcpEventHandler) BeforeStreamStop() {
+	if h.isFinite {
+		return
+	}
 	h.bulk.PrepareStartRebalancing()
 }
 

elasticsearch/bulk/bulk.go

@@ -425,7 +425,7 @@ func joinErrors(body map[string]any) (map[string]string, error) {
 			}
 		}
 	}
-	return ivd, fmt.Errorf(sb.String())
+	return ivd, fmt.Errorf("%s", sb.String())
 }
 
 func (b *Bulk) getIndexName(collectionName, actionIndexName string) string {
@@ -471,7 +471,7 @@ func (b *Bulk) finalizeProcess(batchActions []*document.ESActionDocument, errorD
 			if b.sinkResponseHandler != nil {
 				b.sinkResponseHandler.OnError(&dcpElasticsearch.SinkResponseHandlerContext{
 					Action: action,
-					Err:    fmt.Errorf(errorData[key]),
+					Err:    fmt.Errorf("%s", errorData[key]),
 				})
 			}
 		} else {
@@ -489,9 +489,10 @@ func (b *Bulk) countError(action *document.ESActionDocument) {
 	b.LockMetrics()
 	defer b.UnlockMetrics()
 
-	if action.Type == document.Index || action.Type == document.DocUpdate || action.Type == document.ScriptUpdate {
+	switch action.Type {
+	case document.Index, document.DocUpdate, document.ScriptUpdate:
 		b.metric.IndexingErrorActionCounter[action.IndexName]++
-	} else if action.Type == document.Delete {
+	case document.Delete:
 		b.metric.DeletionErrorActionCounter[action.IndexName]++
 	}
 }
@@ -500,9 +501,10 @@ func (b *Bulk) countSuccess(action *document.ESActionDocument) {
 	b.LockMetrics()
 	defer b.UnlockMetrics()
 
-	if action.Type == document.Index || action.Type == document.DocUpdate || action.Type == document.ScriptUpdate {
+	switch action.Type {
+	case document.Index, document.DocUpdate, document.ScriptUpdate:
 		b.metric.IndexingSuccessActionCounter[action.IndexName]++
-	} else if action.Type == document.Delete {
+	case document.Delete:
 		b.metric.DeletionSuccessActionCounter[action.IndexName]++
 	}
 }

elasticsearch/client/fasthttp_transport.go

@@ -81,9 +81,9 @@ func (t *transport) copyRequest(dst *fasthttp.Request, src *http.Request) *fasth
 func (t *transport) copyResponse(dst *http.Response, src *fasthttp.Response) *http.Response {
 	dst.StatusCode = src.StatusCode()
 
-	src.Header.VisitAll(func(k, v []byte) {
+	for k, v := range src.Header.All() {
 		dst.Header.Set(string(k), string(v))
-	})
+	}
 
 	// Cast to a string to make a copy seeing as src.Body() won't
 	// be valid after the response is released back to the pool (fasthttp.ReleaseResponse).