Repository: beatlabs/patron (go)

Issue:
Kafka consumer consolidation
## Is your feature request related to a problem? Please describe

Up to this point, we had two consumer implementations, simple and group.
Group used consumer groups to handle the messages whereas simple consumed from each partition independently.
The reason for simple was to be able to handle messages with better support for offset since we could be handling the messages for each partition independently.

The same can be achieved by setting the offset to the latest and using as consumer group always a different unique value per pod e.g. a UUID. Kafka has a garbage collector for unused consumer groups following this [value](https://kafka.apache.org/documentation/#brokerconfigs_offsets.retention.minutes).

## Describe the solution

We should be moving forward and using only group one and making it the only one available.



Pull request description:
Consolidate Kafka consumer
<!--
Thanks for taking precious time for making a PR.

Before creating a pull request, please make sure:
- Your PR solves one problem for which an issue exist and a solution has been discussed
- You have read the guide for contributing
  - See https://github.com/beatlabs/patron/blob/master/README.md#how-to-contribute
- You signed all your commits (otherwise we won't be able to merge the PR)
  - See https://github.com/beatlabs/patron/blob/master/SIGNYOURWORK.md
- You added unit tests for the new functionality
- You mention in the PR description which issue it is addressing, e.g. "Resolves #123"
-->

## Which problem is this PR solving?

Resolves #605.

## Short description of the changes

Move group into the kafka consumer package.


Patch:
diff --git a/component/kafka/group/component.go b/component/kafka/component.go
similarity index 92%
rename from component/kafka/group/component.go
rename to component/kafka/component.go
index 00fe6dc13..9d20c8924 100644
--- a/component/kafka/group/component.go
+++ b/component/kafka/component.go
@@ -1,5 +1,4 @@
-// Package group provides kafka consumer group component implementation.
-package group
+package kafka
 
 import (
 	"context"
@@ -10,7 +9,6 @@ import (
 	"time"
 
 	"github.com/Shopify/sarama"
-	"github.com/beatlabs/patron/component/kafka"
 	"github.com/beatlabs/patron/correlation"
 	patronErrors "github.com/beatlabs/patron/errors"
 	"github.com/beatlabs/patron/internal/validation"
@@ -36,7 +34,7 @@ const (
 	defaultRetryWait       = 10 * time.Second
 	defaultBatchSize       = 1
 	defaultBatchTimeout    = 100 * time.Millisecond
-	defaultFailureStrategy = kafka.ExitStrategy
+	defaultFailureStrategy = ExitStrategy
 )
 
 var (
@@ -101,7 +99,7 @@ func messageStatusCountInc(status, group, topic string) {
 // The default failure strategy is the ExitStrategy.
 // The default batch size is 1 and the batch timeout is 100ms.
 // The default number of retries is 0 and the retry wait is 0.
-func New(name, group string, brokers, topics []string, proc kafka.BatchProcessorFunc, saramaCfg *sarama.Config, oo ...OptionFunc) (*Component, error) {
+func New(name, group string, brokers, topics []string, proc BatchProcessorFunc, saramaCfg *sarama.Config, oo ...OptionFunc) (*Component, error) {
 	var errs []error
 	if name == "" {
 		errs = append(errs, errors.New("name is required"))
@@ -162,8 +160,8 @@ type Component struct {
 	topics                    []string
 	brokers                   []string
 	saramaConfig              *sarama.Config
-	proc                      kafka.BatchProcessorFunc
-	failStrategy              kafka.FailStrategy
+	proc                      BatchProcessorFunc
+	failStrategy              FailStrategy
 	batchSize                 uint
 	batchTimeout              time.Duration
 	batchMessageDeduplication bool
@@ -270,10 +268,10 @@ type consumerHandler struct {
 	batchMessageDeduplication bool
 
 	// callback
-	proc kafka.BatchProcessorFunc
+	proc BatchProcessorFunc
 
 	// failures strategy
-	failStrategy kafka.FailStrategy
+	failStrategy FailStrategy
 
 	// committing after every batch
 	commitSync bool
@@ -290,8 +288,8 @@ type consumerHandler struct {
 	sessionCallback   func(sarama.ConsumerGroupSession) error
 }
 
-func newConsumerHandler(ctx context.Context, name, group string, processorFunc kafka.BatchProcessorFunc,
-	fs kafka.FailStrategy, batchSize uint, batchTimeout time.Duration, commitSync, batchMessageDeduplication bool,
+func newConsumerHandler(ctx context.Context, name, group string, processorFunc BatchProcessorFunc,
+	fs FailStrategy, batchSize uint, batchTimeout time.Duration, commitSync, batchMessageDeduplication bool,
 	sessionCallback func(sarama.ConsumerGroupSession) error,
 ) *consumerHandler {
 	return &consumerHandler{
@@ -362,17 +360,17 @@ func (c *consumerHandler) flush(session sarama.ConsumerGroupSession) error {
 		return nil
 	}
 
-	messages := make([]kafka.Message, 0, len(c.msgBuf))
+	messages := make([]Message, 0, len(c.msgBuf))
 	for _, msg := range c.msgBuf {
 		messageStatusCountInc(messageProcessed, c.group, msg.Topic)
 		ctx, sp := c.getContextWithCorrelation(msg)
-		messages = append(messages, kafka.NewMessage(ctx, sp, msg))
+		messages = append(messages, NewMessage(ctx, sp, msg))
 	}
 
 	if c.batchMessageDeduplication {
 		messages = deduplicateMessages(messages)
 	}
-	btc := kafka.NewBatch(messages)
+	btc := NewBatch(messages)
 	err := c.proc(btc)
 	if err != nil {
 		if errors.Is(c.ctx.Err(), context.Canceled) {
@@ -399,9 +397,9 @@ func (c *consumerHandler) flush(session sarama.ConsumerGroupSession) error {
 	return nil
 }
 
-func (c *consumerHandler) executeFailureStrategy(messages []kafka.Message, err error) error {
+func (c *consumerHandler) executeFailureStrategy(messages []Message, err error) error {
 	switch c.failStrategy {
-	case kafka.ExitStrategy:
+	case ExitStrategy:
 		for _, m := range messages {
 			trace.SpanError(m.Span())
 			messageStatusCountInc(messageErrored, c.group, m.Message().Topic)
@@ -409,7 +407,7 @@ func (c *consumerHandler) executeFailureStrategy(messages []kafka.Message, err e
 		slog.Error("could not process message(s)")
 		c.err = err
 		return err
-	case kafka.SkipStrategy:
+	case SkipStrategy:
 		for _, m := range messages {
 			trace.SpanError(m.Span())
 			messageStatusCountInc(messageErrored, c.group, m.Message().Topic)
@@ -468,13 +466,13 @@ func mapHeader(hh []*sarama.RecordHeader) map[string]string {
 // This function assumes that messages are ordered from old to new, and relies on Kafka ordering guarantees within
 // partitions. This is the default behaviour from Kafka unless the Producer altered the partition hashing behaviour in
 // a nondeterministic way.
-func deduplicateMessages(messages []kafka.Message) []kafka.Message {
-	m := map[string]kafka.Message{}
+func deduplicateMessages(messages []Message) []Message {
+	m := map[string]Message{}
 	for _, message := range messages {
 		m[string(message.Message().Key)] = message
 	}
 
-	deduplicated := make([]kafka.Message, 0, len(m))
+	deduplicated := make([]Message, 0, len(m))
 	for _, message := range m {
 		deduplicated = append(deduplicated, message)
 	}
diff --git a/component/kafka/group/option.go b/component/kafka/option.go
similarity index 89%
rename from component/kafka/group/option.go
rename to component/kafka/option.go
index 63db6c9d0..f845d2975 100644
--- a/component/kafka/group/option.go
+++ b/component/kafka/option.go
@@ -1,4 +1,4 @@
-package group
+package kafka
 
 import (
 	"errors"
@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/Shopify/sarama"
-	"github.com/beatlabs/patron/component/kafka"
 	"golang.org/x/exp/slog"
 )
 
@@ -14,13 +13,13 @@ import (
 type OptionFunc func(*Component) error
 
 // WithFailureStrategy sets the strategy to follow for the component when it encounters an error.
-// The kafka.ExitStrategy will fail the component, if there are retries > 0 then the component will reconnect and retry
+// The ExitStrategy will fail the component, if there are retries > 0 then the component will reconnect and retry
 // the failed message.
-// The kafka.SkipStrategy will skip the message on failure. If a client wants to retry a message before failing then
-// this needs to be handled in the kafka.BatchProcessorFunc.
-func WithFailureStrategy(fs kafka.FailStrategy) OptionFunc {
+// The SkipStrategy will skip the message on failure. If a client wants to retry a message before failing then
+// this needs to be handled in the BatchProcessorFunc.
+func WithFailureStrategy(fs FailStrategy) OptionFunc {
 	return func(c *Component) error {
-		if fs > kafka.SkipStrategy || fs < kafka.ExitStrategy {
+		if fs > SkipStrategy || fs < ExitStrategy {
 			return errors.New("invalid failure strategy provided")
 		}
 		c.failStrategy = fs
diff --git a/examples/service/kafka.go b/examples/service/kafka.go
index f0ffc1949..e9c5bdd83 100644
--- a/examples/service/kafka.go
+++ b/examples/service/kafka.go
@@ -6,7 +6,6 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/beatlabs/patron"
 	"github.com/beatlabs/patron/component/kafka"
-	"github.com/beatlabs/patron/component/kafka/group"
 	"github.com/beatlabs/patron/examples"
 	"github.com/beatlabs/patron/log"
 )
@@ -26,7 +25,7 @@ func createKafkaConsumer() (patron.Component, error) {
 		return nil
 	}
 
-	return group.New(name, examples.KafkaGroup, []string{examples.KafkaBroker}, []string{examples.KafkaTopic}, process, cfg,
-		group.WithFailureStrategy(kafka.SkipStrategy), group.WithBatchSize(1), group.WithBatchTimeout(1*time.Second),
-		group.WithRetries(10), group.WithRetryWait(3*time.Second), group.WithCommitSync())
+	return kafka.New(name, examples.KafkaGroup, []string{examples.KafkaBroker}, []string{examples.KafkaTopic}, process, cfg,
+		kafka.WithFailureStrategy(kafka.SkipStrategy), kafka.WithBatchSize(1), kafka.WithBatchTimeout(1*time.Second),
+		kafka.WithRetries(10), kafka.WithRetryWait(3*time.Second), kafka.WithCommitSync())
 }


Test patch:
diff --git a/component/kafka/group/component_test.go b/component/kafka/component_test.go
similarity index 90%
rename from component/kafka/group/component_test.go
rename to component/kafka/component_test.go
index 83432ae66..35d7e6ac8 100644
--- a/component/kafka/group/component_test.go
+++ b/component/kafka/component_test.go
@@ -1,4 +1,4 @@
-package group
+package kafka
 
 import (
 	"context"
@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/Shopify/sarama"
-	"github.com/beatlabs/patron/component/kafka"
 	"github.com/beatlabs/patron/correlation"
 	"github.com/beatlabs/patron/encoding"
 	"github.com/beatlabs/patron/encoding/json"
@@ -34,8 +33,8 @@ func TestNew(t *testing.T) {
 		group        string
 		brokers      []string
 		topics       []string
-		p            kafka.BatchProcessorFunc
-		fs           kafka.FailStrategy
+		p            BatchProcessorFunc
+		fs           FailStrategy
 		retries      uint
 		retryWait    time.Duration
 		batchSize    uint
@@ -49,37 +48,37 @@ func TestNew(t *testing.T) {
 	}{
 		{
 			name:    "success",
-			args:    args{name: "name", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, retryWait: 2, batchTimeout: time.Second, fs: kafka.ExitStrategy, saramaCfg: saramaCfg},
+			args:    args{name: "name", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, retryWait: 2, batchTimeout: time.Second, fs: ExitStrategy, saramaCfg: saramaCfg},
 			wantErr: false,
 		},
 		{
 			name:    "failed, no sarama config",
-			args:    args{name: "name", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, retryWait: 2, batchTimeout: time.Second, fs: kafka.ExitStrategy, saramaCfg: nil},
+			args:    args{name: "name", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, retryWait: 2, batchTimeout: time.Second, fs: ExitStrategy, saramaCfg: nil},
 			wantErr: true,
 		},
 		{
 			name:    "failed, missing name",
-			args:    args{name: "", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, batchTimeout: time.Second, fs: kafka.ExitStrategy, saramaCfg: saramaCfg},
+			args:    args{name: "", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, batchTimeout: time.Second, fs: ExitStrategy, saramaCfg: saramaCfg},
 			wantErr: true,
 		},
 		{
 			name:    "failed, missing group",
-			args:    args{name: "name", group: "", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, batchTimeout: time.Second, fs: kafka.ExitStrategy, saramaCfg: saramaCfg},
+			args:    args{name: "name", group: "", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, batchTimeout: time.Second, fs: ExitStrategy, saramaCfg: saramaCfg},
 			wantErr: true,
 		},
 		{
 			name:    "failed, no brokers",
-			args:    args{name: "name", group: "grp", brokers: []string{}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, batchTimeout: time.Second, fs: kafka.ExitStrategy, saramaCfg: saramaCfg},
+			args:    args{name: "name", group: "grp", brokers: []string{}, topics: []string{"topicone"}, p: proc.Process, batchSize: 1, batchTimeout: time.Second, fs: ExitStrategy, saramaCfg: saramaCfg},
 			wantErr: true,
 		},
 		{
 			name:    "failed, no topics",
-			args:    args{name: "name", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{""}, p: proc.Process, batchSize: 1, batchTimeout: time.Second, fs: kafka.ExitStrategy, saramaCfg: saramaCfg},
+			args:    args{name: "name", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{""}, p: proc.Process, batchSize: 1, batchTimeout: time.Second, fs: ExitStrategy, saramaCfg: saramaCfg},
 			wantErr: true,
 		},
 		{
 			name:    "failed, missing processor func",
-			args:    args{name: "name", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, batchSize: 1, batchTimeout: time.Second, p: nil, fs: kafka.ExitStrategy, saramaCfg: saramaCfg},
+			args:    args{name: "name", group: "grp", brokers: []string{"localhost:9092"}, topics: []string{"topicone"}, batchSize: 1, batchTimeout: time.Second, p: nil, fs: ExitStrategy, saramaCfg: saramaCfg},
 			wantErr: true,
 		},
 		{
@@ -145,7 +144,7 @@ type mockProcessor struct {
 
 var errProcess = errors.New("PROC ERROR")
 
-func (mp *mockProcessor) Process(batch kafka.Batch) error {
+func (mp *mockProcessor) Process(batch Batch) error {
 	mp.mux.Lock()
 	mp.execs += len(batch.Messages())
 	mp.mux.Unlock()
@@ -217,7 +216,7 @@ func TestHandler_ConsumeClaim(t *testing.T) {
 		name                      string
 		msgs                      []*sarama.ConsumerMessage
 		proc                      *mockProcessor
-		failStrategy              kafka.FailStrategy
+		failStrategy              FailStrategy
 		batchSize                 uint
 		batchMessageDeduplication bool
 		expectError               bool
@@ -229,7 +228,7 @@ func TestHandler_ConsumeClaim(t *testing.T) {
 			proc: &mockProcessor{
 				errReturn: false,
 			},
-			failStrategy: kafka.ExitStrategy,
+			failStrategy: ExitStrategy,
 			batchSize:    1,
 			expectError:  false,
 		},
@@ -252,7 +251,7 @@ func TestHandler_ConsumeClaim(t *testing.T) {
 			proc: &mockProcessor{
 				errReturn: false,
 			},
-			failStrategy: kafka.ExitStrategy,
+			failStrategy: ExitStrategy,
 			batchSize:    10,
 			expectError:  false,
 		},
@@ -262,7 +261,7 @@ func TestHandler_ConsumeClaim(t *testing.T) {
 			proc: &mockProcessor{
 				errReturn: true,
 			},
-			failStrategy: kafka.ExitStrategy,
+			failStrategy: ExitStrategy,
 			batchSize:    1,
 			expectError:  true,
 		},
@@ -272,7 +271,7 @@ func TestHandler_ConsumeClaim(t *testing.T) {
 			proc: &mockProcessor{
 				errReturn: true,
 			},
-			failStrategy: kafka.SkipStrategy,
+			failStrategy: SkipStrategy,
 			batchSize:    1,
 			expectError:  false,
 		},
@@ -295,7 +294,7 @@ func TestHandler_ConsumeClaim(t *testing.T) {
 			proc: &mockProcessor{
 				errReturn: false,
 			},
-			failStrategy:              kafka.SkipStrategy,
+			failStrategy:              SkipStrategy,
 			batchSize:                 10,
 			batchMessageDeduplication: true,
 			expectError:               false,
@@ -385,13 +384,13 @@ func Test_getCorrelationID(t *testing.T) {
 }
 
 func Test_deduplicateMessages(t *testing.T) {
-	message := func(key, val string) kafka.Message {
-		return kafka.NewMessage(
+	message := func(key, val string) Message {
+		return NewMessage(
 			context.Background(),
 			opentracing.SpanFromContext(context.Background()),
 			&sarama.ConsumerMessage{Key: []byte(key), Value: []byte(val)})
 	}
-	find := func(collection []kafka.Message, key string) kafka.Message {
+	find := func(collection []Message, key string) Message {
 		for _, m := range collection {
 			if string(m.Message().Key) == key {
 				return m
@@ -401,7 +400,7 @@ func Test_deduplicateMessages(t *testing.T) {
 	}
 
 	// Given
-	original := []kafka.Message{
+	original := []Message{
 		message("k1", "v1.1"),
 		message("k1", "v1.2"),
 		message("k2", "v2.1"),
diff --git a/component/kafka/group/integration_test.go b/component/kafka/integration_test.go
similarity index 93%
rename from component/kafka/group/integration_test.go
rename to component/kafka/integration_test.go
index 64d1213dd..6aa631bcf 100644
--- a/component/kafka/group/integration_test.go
+++ b/component/kafka/integration_test.go
@@ -1,7 +1,7 @@
 //go:build integration
 // +build integration
 
-package group
+package kafka
 
 import (
 	"context"
@@ -14,7 +14,6 @@ import (
 
 	"github.com/Shopify/sarama"
 	kafkaclient "github.com/beatlabs/patron/client/kafka"
-	"github.com/beatlabs/patron/component/kafka"
 	"github.com/beatlabs/patron/correlation"
 	testkafka "github.com/beatlabs/patron/test/kafka"
 	"github.com/opentracing/opentracing-go"
@@ -64,7 +63,7 @@ func TestKafkaComponent_Success(t *testing.T) {
 	actualSuccessfulMessages := make([]string, 0)
 	var consumerWG sync.WaitGroup
 	consumerWG.Add(numOfMessagesToSend)
-	processorFunc := func(batch kafka.Batch) error {
+	processorFunc := func(batch Batch) error {
 		for _, msg := range batch.Messages() {
 			var msgContent string
 			err := decodeString(msg.Message().Value, &msgContent)
@@ -128,7 +127,7 @@ func TestKafkaComponent_FailAllRetries(t *testing.T) {
 	// Set up the kafka component
 	actualSuccessfulMessages := make([]int, 0)
 	actualNumOfRuns := int32(0)
-	processorFunc := func(batch kafka.Batch) error {
+	processorFunc := func(batch Batch) error {
 		for _, msg := range batch.Messages() {
 			var msgContent string
 			err := decodeString(msg.Message().Value, &msgContent)
@@ -184,7 +183,7 @@ func TestKafkaComponent_FailOnceAndRetry(t *testing.T) {
 	actualMessages := make([]string, 0)
 	var consumerWG sync.WaitGroup
 	consumerWG.Add(numOfMessagesToSend)
-	processorFunc := func(batch kafka.Batch) error {
+	processorFunc := func(batch Batch) error {
 		for _, msg := range batch.Messages() {
 			var msgContent string
 			err := decodeString(msg.Message().Value, &msgContent)
@@ -240,7 +239,7 @@ func TestKafkaComponent_FailOnceAndRetry(t *testing.T) {
 
 func TestGroupConsume_CheckTopicFailsDueToNonExistingTopic(t *testing.T) {
 	// Test parameters
-	processorFunc := func(batch kafka.Batch) error {
+	processorFunc := func(batch Batch) error {
 		return nil
 	}
 	invalidTopicName := "invalid-topic-name"
@@ -251,7 +250,7 @@ func TestGroupConsume_CheckTopicFailsDueToNonExistingTopic(t *testing.T) {
 
 func TestGroupConsume_CheckTopicFailsDueToNonExistingBroker(t *testing.T) {
 	// Test parameters
-	processorFunc := func(batch kafka.Batch) error {
+	processorFunc := func(batch Batch) error {
 		return nil
 	}
 	_, err := New(successTopic2, successTopic2+"-group", []string{"127.0.0.1:9999"},
@@ -260,14 +259,14 @@ func TestGroupConsume_CheckTopicFailsDueToNonExistingBroker(t *testing.T) {
 	require.Contains(t, err.Error(), "failed to create client:")
 }
 
-func newComponent(t *testing.T, name string, retries uint, batchSize uint, processorFunc kafka.BatchProcessorFunc) *Component {
-	saramaCfg, err := kafka.DefaultConsumerSaramaConfig(name, true)
+func newComponent(t *testing.T, name string, retries uint, batchSize uint, processorFunc BatchProcessorFunc) *Component {
+	saramaCfg, err := DefaultConsumerSaramaConfig(name, true)
 	saramaCfg.Consumer.Offsets.Initial = sarama.OffsetOldest
 	saramaCfg.Version = sarama.V2_6_0_0
 	require.NoError(t, err)
 
 	cmp, err := New(name, name+"-group", []string{broker}, []string{name}, processorFunc,
-		saramaCfg, WithFailureStrategy(kafka.ExitStrategy), WithBatchSize(batchSize), WithBatchTimeout(100*time.Millisecond),
+		saramaCfg, WithFailureStrategy(ExitStrategy), WithBatchSize(batchSize), WithBatchTimeout(100*time.Millisecond),
 		WithRetries(retries), WithRetryWait(200*time.Millisecond), WithCommitSync(), WithCheckTopic())
 	require.NoError(t, err)
 
diff --git a/component/kafka/kafka_test.go b/component/kafka/kafka_test.go
index 8d5f42ab0..d534b5171 100644
--- a/component/kafka/kafka_test.go
+++ b/component/kafka/kafka_test.go
@@ -6,12 +6,11 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/opentracing/opentracing-go/mocktracer"
-	"github.com/stretchr/testify/require"
-
 	"github.com/Shopify/sarama"
 	"github.com/beatlabs/patron/correlation"
+	"github.com/opentracing/opentracing-go/mocktracer"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func Test_messageWrapper(t *testing.T) {
diff --git a/component/kafka/group/option_test.go b/component/kafka/option_test.go
similarity index 95%
rename from component/kafka/group/option_test.go
rename to component/kafka/option_test.go
index 4233644de..3cbeeac80 100644
--- a/component/kafka/group/option_test.go
+++ b/component/kafka/option_test.go
@@ -1,28 +1,27 @@
-package group
+package kafka
 
 import (
 	"testing"
 	"time"
 
 	"github.com/Shopify/sarama"
-	"github.com/beatlabs/patron/component/kafka"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestFailureStrategy(t *testing.T) {
 	t.Parallel()
 	type args struct {
-		strategy kafka.FailStrategy
+		strategy FailStrategy
 	}
 	tests := map[string]struct {
 		args        args
 		expectedErr string
 	}{
 		"success-exit": {
-			args: args{strategy: kafka.ExitStrategy},
+			args: args{strategy: ExitStrategy},
 		},
 		"success-skip": {
-			args: args{strategy: kafka.SkipStrategy},
+			args: args{strategy: SkipStrategy},
 		},
 		"invalid strategy": {
 			args:        args{strategy: -1},