feat(kafka): support transaction isolation levels

internal/impl/kafka/input_kafka_franz.go

@@ -77,6 +77,7 @@ Finally, it's also possible to specify an explicit offset to consume from by add
 		Field(service.NewStringField("consumer_group").
 			Description("An optional consumer group to consume as. When specified the partitions of specified topics are automatically distributed across consumers sharing a consumer group, and partition offsets are automatically committed and resumed under this name. Consumer groups are not supported when specifying explicit partitions to consume from in the `topics` field.").
 			Optional()).
+		Field(newTransactionIsolationLevelField()).
 		Field(service.NewStringField("client_id").
 			Description("An identifier for the client connection.").
 			Default("bento").
@@ -207,6 +208,7 @@ type franzKafkaReader struct {
 	saslConfs       []sasl.Mechanism
 	checkpointLimit int
 	autoOffsetReset string
+	isolationLevel  kgo.IsolationLevel
 	commitPeriod    time.Duration
 	regexPattern    bool
 	multiHeader     bool
@@ -361,6 +363,17 @@ func newFranzKafkaReaderFromConfig(conf *service.ParsedConfig, res *service.Reso
 		return nil, err
 	}
 
+	isolationLevel, err := transactionIsolationLevelFromConfig(conf)
+	if err != nil {
+		return nil, err
+	}
+	switch isolationLevel {
+	case transactionIsolationLevelReadUncommitted:
+		f.isolationLevel = kgo.ReadUncommitted()
+	case transactionIsolationLevelReadCommitted:
+		f.isolationLevel = kgo.ReadCommitted()
+	}
+
 	if f.reconnectOnUnknownTopic, err = conf.FieldBool("reconnect_on_unknown_topic_or_partition"); err != nil {
 		return nil, err
 	}
@@ -851,6 +864,7 @@ func (f *franzKafkaReader) Connect(ctx context.Context) error {
 		kgo.ConsumeTopics(f.topics...),
 		kgo.ConsumePartitions(f.topicPartitions),
 		kgo.ConsumeResetOffset(initialOffset),
+		kgo.FetchIsolationLevel(f.isolationLevel),
 		kgo.SASL(f.saslConfs...),
 		kgo.ConsumerGroup(f.consumerGroup),
 		kgo.ClientID(f.clientID),

internal/impl/kafka/input_kafka_franz_test.go

@@ -85,6 +85,59 @@ topic: foo
 	}
 }
 
+func TestInputKafkaFranzTransactionIsolationLevel(t *testing.T) {
+	testCases := []struct {
+		name     string
+		config   string
+		expected kgo.IsolationLevel
+	}{
+		{
+			name:     "defaults to read uncommitted",
+			expected: kgo.ReadUncommitted(),
+		},
+		{
+			name:     "explicit read uncommitted",
+			config:   "transaction_isolation_level: read_uncommitted",
+			expected: kgo.ReadUncommitted(),
+		},
+		{
+			name:     "read committed",
+			config:   "transaction_isolation_level: read_committed",
+			expected: kgo.ReadCommitted(),
+		},
+	}
+
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+			conf, err := franzKafkaInputConfig().ParseYAML(fmt.Sprintf(`
+seed_brokers: [ example.com:1234 ]
+topics: [ test ]
+consumer_group: test
+%s
+`, test.config), nil)
+			require.NoError(t, err)
+
+			reader, err := newFranzKafkaReaderFromConfig(conf, service.MockResources())
+			require.NoError(t, err)
+			assert.Equal(t, test.expected, reader.isolationLevel)
+		})
+	}
+}
+
+func TestInputKafkaFranzRejectsInvalidTransactionIsolationLevel(t *testing.T) {
+	conf, err := franzKafkaInputConfig().ParseYAML(`
+seed_brokers: [ example.com:1234 ]
+topics: [ test ]
+consumer_group: test
+transaction_isolation_level: eventually_consistent
+`, nil)
+	require.NoError(t, err)
+
+	_, err = newFranzKafkaReaderFromConfig(conf, service.MockResources())
+	require.Error(t, err)
+	assert.EqualError(t, err, `invalid transaction isolation level: "eventually_consistent"`)
+}
+
 func TestInputKafkaFranzRetriableError(t *testing.T) {
 	conf, err := franzKafkaInputConfig().ParseYAML(`
 seed_brokers: [ localhost:9092 ]

internal/impl/kafka/input_sarama_kafka.go

@@ -131,6 +131,7 @@ Unfortunately this error message will appear for a wide range of connection prob
 			service.NewStringField(iskFieldConsumerGroup).
 				Description("An identifier for the consumer group of the connection. This field can be explicitly made empty in order to disable stored offsets for the consumed topic partitions.").
 				Default(""),
+			newTransactionIsolationLevelField(),
 			service.NewStringField(iskFieldClientID).
 				Description("An identifier for the client connection.").
 				Advanced().Default("bento"),
@@ -480,6 +481,16 @@ func (k *kafkaReader) saramaConfigFromParsed(conf *service.ParsedConfig) (*saram
 
 	config.Net.DialTimeout = time.Second
 	config.Consumer.Return.Errors = true
+	isolationLevel, err := transactionIsolationLevelFromConfig(conf)
+	if err != nil {
+		return nil, err
+	}
+	switch isolationLevel {
+	case transactionIsolationLevelReadUncommitted:
+		config.Consumer.IsolationLevel = sarama.ReadUncommitted
+	case transactionIsolationLevelReadCommitted:
+		config.Consumer.IsolationLevel = sarama.ReadCommitted
+	}
 	if config.Consumer.MaxProcessingTime, err = conf.FieldDuration(iskFieldMaxProcessingPeriod); err != nil {
 		return nil, err
 	}

internal/impl/kafka/input_sarama_kafka_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/IBM/sarama"
 	"github.com/Jeffail/gabs/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -48,3 +49,56 @@ topics: %v
 		})
 	}
 }
+
+func TestKafkaTransactionIsolationLevel(t *testing.T) {
+	testCases := []struct {
+		name     string
+		config   string
+		expected sarama.IsolationLevel
+	}{
+		{
+			name:     "defaults to read uncommitted",
+			expected: sarama.ReadUncommitted,
+		},
+		{
+			name:     "explicit read uncommitted",
+			config:   "transaction_isolation_level: read_uncommitted",
+			expected: sarama.ReadUncommitted,
+		},
+		{
+			name:     "read committed",
+			config:   "transaction_isolation_level: read_committed",
+			expected: sarama.ReadCommitted,
+		},
+	}
+
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+			conf, err := iskConfigSpec().ParseYAML(fmt.Sprintf(`
+addresses: [ example.com:1234 ]
+topics: [ test ]
+consumer_group: test
+%s
+`, test.config), nil)
+			require.NoError(t, err)
+
+			reader, err := newKafkaReaderFromParsed(conf, service.MockResources())
+			require.NoError(t, err)
+			assert.Equal(t, test.expected, reader.saramConf.Consumer.IsolationLevel)
+		})
+	}
+}
+
+func TestKafkaRejectsInvalidTransactionIsolationLevel(t *testing.T) {
+	conf, err := iskConfigSpec().ParseYAML(`
+addresses: [ example.com:1234 ]
+topics: [ test ]
+consumer_group: test
+transaction_isolation_level: eventually_consistent
+`, nil)
+	require.NoError(t, err)
+
+	_, err = newKafkaReaderFromParsed(conf, service.MockResources())
+	require.Error(t, err)
+	assert.EqualError(t, err, `invalid transaction isolation level: "eventually_consistent"`)
+}