Function TestTxnOffsetCommitGroupMetadata

transaction_manager_test.go:467–517

(t *testing.T)

}

func TestTxnOffsetCommitGroupMetadata(t *testing.T) {
	broker := NewMockBroker(t, 1)
	defer broker.Close()

	config := NewTestConfig()
	config.Producer.Idempotent = true
	config.Producer.Transaction.ID = "test"
	config.Version = V2_5_0_0
	config.Producer.RequiredAcks = WaitForAll
	config.Net.MaxOpenRequests = 1

	client, txmng := newMockTxnManager(t, broker, config)
	defer client.Close()
	txmng.status = ProducerTxnFlagInTransaction

	broker.Returns(&AddOffsetsToTxnResponse{Err: ErrNoError})
	broker.Returns(&FindCoordinatorResponse{Coordinator: client.Brokers()[0], Err: ErrNoError})
	broker.Returns(&TxnOffsetCommitResponse{
		Topics: map[string][]*PartitionError{
			"test-topic": {{Partition: 0, Err: ErrNoError}},
		},
	})

	instanceID := "instance-1"
	groupMetadata := &ConsumerGroupMetadata{
		GroupID:         "test-group",
		GenerationID:    42,
		MemberID:        "member-1",
		GroupInstanceID: &instanceID,
	}
	offsets := topicPartitionOffsets{
		topicPartition{topic: "test-topic", partition: 0}: {Partition: 0, Offset: 0},
	}

	_, err := txmng.publishOffsetsToTxn(offsets, groupMetadata)
	require.NoError(t, err)

	var committed *TxnOffsetCommitRequest
	for _, rr := range broker.History() {
		if req, ok := rr.Request.(*TxnOffsetCommitRequest); ok {
			committed = req
		}
	}
	require.NotNil(t, committed, "expected a TxnOffsetCommitRequest to reach the broker")
	assert.EqualValues(t, 3, committed.Version)
	assert.Equal(t, "test-group", committed.GroupID)
	assert.EqualValues(t, 42, committed.GenerationID)
	assert.Equal(t, "member-1", committed.MemberID)
	require.NotNil(t, committed.GroupInstanceID)
	assert.Equal(t, instanceID, *committed.GroupInstanceID)
}

func TestTxnOffsetsCommit(t *testing.T) {
	type testCase struct {