Method InitProducerID

initproducerid.go:61–82

InitProducerID sends a initProducerId request to a kafka broker and returns the response.

(ctx context.Context, req *InitProducerIDRequest)

// InitProducerID sends a initProducerId request to a kafka broker and returns the
// response.
func (c *Client) InitProducerID(ctx context.Context, req *InitProducerIDRequest) (*InitProducerIDResponse, error) {
	m, err := c.roundTrip(ctx, req.Addr, &initproducerid.Request{
		TransactionalID:      req.TransactionalID,
		TransactionTimeoutMs: int32(req.TransactionTimeoutMs),
		ProducerID:           int64(req.ProducerID),
		ProducerEpoch:        int16(req.ProducerEpoch),
	})
	if err != nil {
		return nil, fmt.Errorf("kafka.(*Client).InitProducerId: %w", err)
	}

	res := m.(*initproducerid.Response)

	return &InitProducerIDResponse{
		Producer: &ProducerSession{
			ProducerID:    int(res.ProducerID),
			ProducerEpoch: int(res.ProducerEpoch),
		},
		Throttle: makeDuration(res.ThrottleTimeMs),
		Error:    makeError(res.ErrorCode, ""),
	}, nil
}