Function TestFuncTxnProduceAndCommitOffset

functional_producer_test.go:385–415

(t *testing.T)

}

func TestFuncTxnProduceAndCommitOffset(t *testing.T) {
	checkKafkaVersion(t, "0.11.0.0")

	t.Run("legacy group ID", func(t *testing.T) {
		testTxnProduceAndCommitOffset(t, MaxVersion, "TestFuncTxnProduceAndCommitOffset", "test-produce",
			func(producer AsyncProducer, _ ConsumerGroupSession, msg *ConsumerMessage, groupID string) error {
				return producer.AddMessageToTxn(msg, groupID, nil)
			})
	})

	// test the v2/v3 TxnOffsetCommit protocol boundaries across broker versions
	versions := []struct {
		name     string
		version  KafkaVersion
		minKafka string
	}{
		{"v2", V2_1_0_0, "2.1.0"},
		{"v3 with group metadata", V2_5_0_0, "2.5.0"},
	}
	for _, v := range versions {
		t.Run(v.name, func(t *testing.T) {
			checkKafkaVersion(t, v.minKafka)
			groupID := "test-produce-" + v.version.String()
			txnID := "TestFuncTxnProduceAndCommitOffset-" + v.version.String()
			testTxnProduceAndCommitOffset(t, v.version, txnID, groupID,
				func(producer AsyncProducer, sess ConsumerGroupSession, msg *ConsumerMessage, groupID string) error {
					return producer.AddMessageToTxnWithGroupMetadata(msg, NewConsumerGroupMetadataFromSession(sess, groupID, nil), nil)
				})
		})
	}
}

func testTxnProduceAndCommitOffset(
	t *testing.T,