Method CreateTopics

createtopics.go:49–82

CreateTopics sends a topic creation request to a kafka broker and returns the response.

(ctx context.Context, req *CreateTopicsRequest)

// CreateTopics sends a topic creation request to a kafka broker and returns the
// response.
func (c *Client) CreateTopics(ctx context.Context, req *CreateTopicsRequest) (*CreateTopicsResponse, error) {
	topics := make([]createtopics.RequestTopic, len(req.Topics))

	for i, t := range req.Topics {
		topics[i] = createtopics.RequestTopic{
			Name:              t.Topic,
			NumPartitions:     int32(t.NumPartitions),
			ReplicationFactor: int16(t.ReplicationFactor),
			Assignments:       t.assignments(),
			Configs:           t.configs(),
		}
	}

	m, err := c.roundTrip(ctx, req.Addr, &createtopics.Request{
		Topics:       topics,
		TimeoutMs:    c.timeoutMs(ctx, defaultCreateTopicsTimeout),
		ValidateOnly: req.ValidateOnly,
	})
	if err != nil {
		return nil, fmt.Errorf("kafka.(*Client).CreateTopics: %w", err)
	}

	res := m.(*createtopics.Response)
	ret := &CreateTopicsResponse{
		Throttle: makeDuration(res.ThrottleTimeMs),
		Errors:   make(map[string]error, len(res.Topics)),
	}

	for _, t := range res.Topics {
		ret.Errors[t.Name] = makeError(t.ErrorCode, t.ErrorMessage)
	}

	return ret, nil
}

type ConfigEntry struct {
	ConfigName  string