Function TestTxnOffsetsCommit

transaction_manager_test.go:519–727

(t *testing.T)

}

func TestTxnOffsetsCommit(t *testing.T) {
	type testCase struct {
		brokerErr       KError
		initialFlags    ProducerTxnStatusFlag
		initialOffsets  topicPartitionOffsets
		expectedFlags   ProducerTxnStatusFlag
		expectedError   error
		expectedOffsets topicPartitionOffsets
	}

	originalOffsets := topicPartitionOffsets{
		topicPartition{topic: "test-topic", partition: 0}: {
			Partition: 0,
			Offset:    0,
		},
	}

	testCases := []testCase{
		{
			brokerErr:    ErrConsumerCoordinatorNotAvailable,
			initialFlags: ProducerTxnFlagInTransaction,
			initialOffsets: topicPartitionOffsets{
				topicPartition{topic: "test-topic", partition: 0}: {
					Partition: 0,
					Offset:    0,
				},
			},
			expectedFlags:   ProducerTxnFlagInTransaction,
			expectedError:   Wrap(ErrTxnOffsetCommit, ErrConsumerCoordinatorNotAvailable),
			expectedOffsets: originalOffsets,
		},
		{
			brokerErr:    ErrNotCoordinatorForConsumer,
			initialFlags: ProducerTxnFlagInTransaction,
			initialOffsets: topicPartitionOffsets{
				topicPartition{topic: "test-topic", partition: 0}: {
					Partition: 0,
					Offset:    0,
				},
			},
			expectedFlags:   ProducerTxnFlagInTransaction,
			expectedError:   Wrap(ErrTxnOffsetCommit, ErrNotCoordinatorForConsumer),
			expectedOffsets: originalOffsets,
		},
		{
			brokerErr:    ErrNoError,
			initialFlags: ProducerTxnFlagInTransaction,
			initialOffsets: topicPartitionOffsets{
				topicPartition{topic: "test-topic", partition: 0}: {
					Partition: 0,
					Offset:    0,
				},
			},
			expectedFlags:   ProducerTxnFlagInTransaction,
			expectedError:   nil,
			expectedOffsets: topicPartitionOffsets{},
		},
		{