Function NewKafkaClient

pkg/ingest/testkafka/client.go:16–29

(t testing.TB, address, topic string)

)

func NewKafkaClient(t testing.TB, address, topic string) *kgo.Client {
	writeClient, err := kgo.NewClient(
		kgo.SeedBrokers(address),
		kgo.AllowAutoTopicCreation(),
		kgo.DefaultProduceTopic(topic),
		// We will choose the Partition of each record.
		kgo.RecordPartitioner(kgo.ManualPartitioner()),
		kgo.DisableClientMetrics(),
	)
	require.NoError(t, err)
	t.Cleanup(writeClient.Close)

	return writeClient
}

type ReqOpts struct {
	Partition int32