Added WithWorkTextMapPropagator option (#28)

1. Added `WithWorkTextMapPropagator`
2. Fixed linter in pipeline
3. Addressed linter errors

---------

Co-authored-by: stewartboyd119 <[email protected]>

Author: stewartboyd119

.golangci.yml

@@ -14,31 +14,19 @@ linters-settings:
     sections:
       - standard
       - default
-  depguard:
-    rules:
-      Main:
-        files:
-          - $all
-          - "!$test"
-        deny:
-          - github.com/satori/go.uuid: Prefer "github.com/google/uuid"
+linters:
   disable-all: true
   enable:
     - asciicheck
     - bidichk
     - bodyclose
-    - cyclop
     - decorder
-    - depguard
-    - deadcode
     - dupl
     - errcheck
     - errchkjson
     - errname
     - errorlint
-    - exportloopref
     - gci
-    - gocognit
     - goconst
     - gocritic
     - gocyclo
@@ -49,10 +37,9 @@ linters-settings:
     - nolintlint
     - prealloc
     - staticcheck
-    - structcheck
     - typecheck
     - unconvert
     - unparam
     - unused
-    - varcheck
+    - copyloopvar
     - whitespace

Makefile

@@ -1,6 +1,6 @@
 # Directories containing independent Go modules.
 MODULE_DIRS = .
-GOLANGCI_VERSION=1.64.5
+GOLANGCI_VERSION=1.64.6
 AVRO_CMD_PATH=github.com/hamba/avro/v2/cmd/[email protected]
 SCHEMA_REGISTRY_DOMAIN=schema-registry.shared.zg-int.net:443
 

changelog.md

@@ -4,10 +4,12 @@ All notable changes to this project will be documented in this file.
 
 This project adheres to Semantic Versioning.
 
-## 2.1.2 (March 7, 2025)
+## 2.2.0 (March 9, 2025)
 
-1. Updated module to go1.24
-2. Updated dependencies 
+1. Added `WithWorkTextMapPropagator`
+2. Updated module to go1.24
+3. Updated dependencies 
+4. Fixed linter
 
 ## 2.1.1 (Feb 13, 2025)
 

client.go

@@ -186,8 +186,7 @@ func (c *Client) getFormatter(args formatterArgs) (kFormatter, error) {
 		if err != nil {
 			return nil, err
 		}
-		cf, err := newAvroSchemaRegistryFormatter(scl)
-		return cf, err
+		return newAvroSchemaRegistryFormatter(scl), nil
 	case ProtoSchemaRegistry:
 		scl, err := c.srf.createProto(args.srCfg)
 		if err != nil {

example/worker-delay/main.go

@@ -18,7 +18,7 @@ func main() {
 		BootstrapServers: []string{"localhost:29092"},
 	},
 	// optionally add a logger, which implements zkafka.Logger, to see detailed information about message processsing
-	//zkafka.LoggerOption(),
+	// zkafka.LoggerOption(),
 	)
 	// It's important to close the client after consumption to gracefully leave the consumer group
 	// (this commits completed work, and informs the broker that this consumer is leaving the group which yields a faster rebalance)

example/worker/main.go

@@ -19,7 +19,7 @@ func main() {
 		BootstrapServers: []string{"localhost:29092"},
 	},
 	// optionally add a logger, which implements zkafka.Logger, to see detailed information about message processsing
-	//zkafka.LoggerOption(),
+	// zkafka.LoggerOption(),
 	)
 	// It's important to close the client after consumption to gracefully leave the consumer group
 	// (this commits completed work, and informs the broker that this consumer is leaving the group which yields a faster rebalance)
@@ -79,8 +79,8 @@ func (p Processor) Process(_ context.Context, msg *zkafka.Message) error {
 	// optionally, if you don't want to use the configured formatter at all, access the kafka message payload bytes directly.
 	// The commented out block shows accessing the byte array. In this case we're stringifying the bytes, but this could be json unmarshalled,
 	// proto unmarshalled etc., depending on the expected payload
-	//data := msg.Value()
-	//str := string(data)
+	// data := msg.Value()
+	// str := string(data)
 
 	log.Printf(" offset: %d, partition: %d. event.Name: %s, event.Age %d\n", msg.Offset, msg.Partition, event.Name, event.Age)
 	return nil

formatter.go

@@ -101,10 +101,10 @@ type avroSchemaRegistryFormatter struct {
 	afmt avroFmt
 }
 
-func newAvroSchemaRegistryFormatter(afmt avroFmt) (avroSchemaRegistryFormatter, error) {
+func newAvroSchemaRegistryFormatter(afmt avroFmt) avroSchemaRegistryFormatter {
 	return avroSchemaRegistryFormatter{
 		afmt: afmt,
-	}, nil
+	}
 }
 
 // marshall looks a subject's schema (id) so that it can prefix the eventual message payload.
@@ -170,7 +170,6 @@ func (f avroSchemaRegistryFormatter) unmarshal(req unmarshReq) error {
 	}
 	err = avro.Unmarshal(resolvedSchema, req.data[5:], req.target)
 	if err != nil {
-
 		return fmt.Errorf("failed to deserialize to confluent schema registry avro type: %w", err)
 	}
 	return nil

lifecycle.go

@@ -106,7 +106,6 @@ func ChainLifecycleHooks(hooks ...LifecycleHooks) LifecycleHooks {
 			}
 
 			return hookCtx, allErrs
-
 		},
 		PostReadImmediate: func(ctx context.Context, meta LifecyclePostReadImmediateMeta) {
 			for _, h := range hooks {

reader.go

@@ -115,6 +115,7 @@ func (r *KReader) Read(ctx context.Context) (*Message, error) {
 	kmsg, err := r.consumer.ReadMessage(time.Duration(*r.topicConfig.ReadTimeoutMillis) * time.Millisecond)
 	if err != nil {
 		var v kafka.Error
+		//nolint:gocritic //single switch is okay
 		switch {
 		case errors.As(err, &v):
 			// timeouts occur (because the assigned partitions aren't being written to, lack of activity, etc.). We'll

schemareg.go

@@ -79,7 +79,6 @@ func (c *schemaRegistryFactory) createProto(srConfig SchemaRegistryConfig) (prot
 		ser:   ser,
 		deser: deser,
 	}, nil
-
 }
 
 func (c *schemaRegistryFactory) createJson(srConfig SchemaRegistryConfig) (jsonFmt, error) {

work.go

@@ -161,7 +161,6 @@ func (w *Work) execProcessors(ctx context.Context, shutdown <-chan struct{}) {
 	wg := sync.WaitGroup{}
 	wg.Add(len(w.virtualPartitions))
 	for i := range w.virtualPartitions {
-		i := i
 		go func() {
 			w.processVirtualPartition(ctx, i, shutdown)
 			wg.Done()
@@ -452,6 +451,7 @@ func (w *Work) processSingle(ctx context.Context, msg *Message, partitionIndex i
 	span.RecordError(err)
 	span.SetStatus(codes.Error, err.Error())
 
+	//nolint:errorlint // guaranteed to not be wrapped
 	if pError, ok := err.(processorError); ok {
 		// Because we assume workers will log their own internal errors once
 		// already, we try to ignore logging them twice by also logging them

work_test.go

@@ -78,6 +78,22 @@ func TestWork_WithOptions(t *testing.T) {
 	require.Equal(t, tp.Tracer(""), work.tracer)
 }
 
+func TestWork_WithTextMapPropagator_CanOverrideFactoryPropagator(t *testing.T) {
+	defer recoverThenFail(t)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	propagator := propagation.TraceContext{}
+
+	wf := NewWorkFactory(FakeClient{}, WithTextMapPropagator(propagator))
+
+	w1 := wf.Create(ConsumerTopicConfig{}, &timeDelayProcessor{})
+	require.NotNil(t, w1.p)
+
+	w2 := wf.Create(ConsumerTopicConfig{}, &timeDelayProcessor{}, WithWorkTextMapPropagator(nil))
+	require.Nil(t, w2.p)
+}
+
 // TestWork_ShouldCommitMessagesProperly asserts the behavior of committing kafka messages.
 // Messages should be committed as they complete as long as there aren't lower offset messages still in progress.
 // This tests specifies processing delay times such that low offsets finish after high offsets and asserts that the storeOffsets method

workoption.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"time"
+
+	"go.opentelemetry.io/otel/propagation"
 )
 
 // WorkOption interface to identify functional options
@@ -62,6 +64,11 @@ func WithDeadLetterTopic(deadLetterTopicConfig ProducerTopicConfig) WorkOption {
 	return dltOption{dltConfig: deadLetterTopicConfig}
 }
 
+// WithWorkTextMapPropagator enables the specification of a propagator at the moment a work instance is instantiated
+func WithWorkTextMapPropagator(p propagation.TextMapPropagator) WorkOption {
+	return textMapPropagator{p: p}
+}
+
 type speedupOption struct{ times uint16 }
 
 func (s speedupOption) apply(w *Work) {
@@ -120,6 +127,14 @@ func (d disableBlbOption) apply(w *Work) {
 	w.blb.disabled = d.disabled
 }
 
+type textMapPropagator struct {
+	p propagation.TextMapPropagator
+}
+
+func (d textMapPropagator) apply(w *Work) {
+	w.p = d.p
+}
+
 type dltOption struct {
 	dltConfig ProducerTopicConfig
 }