Function SendReqWithOpts

pkg/ingest/testkafka/client.go:49–63

nolint: revive

(ctx context.Context, t testing.TB, client *kgo.Client, encode encodingFn, opts ReqOpts)

// nolint: revive
func SendReqWithOpts(ctx context.Context, t testing.TB, client *kgo.Client, encode encodingFn, opts ReqOpts) []*kgo.Record {
	traceID := generateTraceID(t)
	opts.applyDefaults()

	startTime := uint64(opts.Time.UnixNano())
	endTime := uint64(opts.Time.Add(time.Second).UnixNano())
	req := test.MakePushBytesRequest(t, 10, traceID, startTime, endTime)
	records, err := encode(opts.Partition, opts.TenantID, req, 1_000_000)
	require.NoError(t, err)

	res := client.ProduceSync(ctx, records...)
	require.NoError(t, res.FirstErr())

	return records
}

func SendReq(ctx context.Context, t testing.TB, client *kgo.Client, encode encodingFn, tenantID string) []*kgo.Record {
	return SendReqWithOpts(ctx, t, client, encode, ReqOpts{Partition: 0, Time: time.Now(), TenantID: tenantID})