Method deleteTopics

deletetopics.go:165–199

deleteTopics deletes the specified topics. See http://kafka.apache.org/protocol.html#The_Messages_DeleteTopics

(request deleteTopicsRequest)

//
// See http://kafka.apache.org/protocol.html#The_Messages_DeleteTopics
func (c *Conn) deleteTopics(request deleteTopicsRequest) (deleteTopicsResponse, error) {
	version, err := c.negotiateVersion(deleteTopics, v0, v1)
	if err != nil {
		return deleteTopicsResponse{}, err
	}

	response := deleteTopicsResponse{
		v: version,
	}

	err = c.writeOperation(
		func(deadline time.Time, id int32) error {
			if request.Timeout == 0 {
				now := time.Now()
				deadline = adjustDeadlineForRTT(deadline, now, defaultRTT)
				request.Timeout = milliseconds(deadlineToTimeout(deadline, now))
			}
			return c.writeRequest(deleteTopics, version, id, request)
		},
		func(deadline time.Time, size int) error {
			return expectZeroSize(func() (remain int, err error) {
				return (&response).readFrom(&c.rbuf, size)
			}())
		},
	)
	if err != nil {
		return deleteTopicsResponse{}, err
	}
	for _, c := range response.TopicErrorCodes {
		if c.ErrorCode != 0 {
			return response, Error(c.ErrorCode)
		}
	}
	return response, nil
}