Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Getting Topic Information when UnknownTopicOrPartition happens #1306

Open
mhmtszr opened this issue Jul 5, 2024 · 0 comments
Open

Getting Topic Information when UnknownTopicOrPartition happens #1306

mhmtszr opened this issue Jul 5, 2024 · 0 comments

Comments

@mhmtszr
Copy link
Contributor

mhmtszr commented Jul 5, 2024

We need to see which topics do not exist in Kafka clusters if we get an error from WriteMessages. We can't enable the "Auto Topic Creation" config because of our business needs. We can't extract the error from "WriteErrors" because error happens before batch operations.

Describe the solution you would like
Like MessageTooLargeError

func messageTooLarge(msgs []Message, i int) MessageTooLargeError {
	remain := make([]Message, 0, len(msgs)-1)
	remain = append(remain, msgs[:i]...)
	remain = append(remain, msgs[i+1:]...)
	return MessageTooLargeError{
		Message:   msgs[i],
		Remaining: remain,
	}
}

We need to implement a custom type error.

func (w *Writer) partitions(ctx context.Context, topic string) (int, error) {
	client := w.client(w.readTimeout())
	// Here we use the transport directly as an optimization to avoid the
	// construction of temporary request and response objects made by the
	// (*Client).Metadata API.
	//
	// It is expected that the transport will optimize this request by
	// caching recent results (the kafka.Transport types does).
	r, err := client.transport().RoundTrip(ctx, client.Addr, &metadataAPI.Request{
		TopicNames:             []string{topic},
		AllowAutoTopicCreation: w.AllowAutoTopicCreation,
	})
	if err != nil {
		return 0, err
	}
	for _, t := range r.(*metadataAPI.Response).Topics {
		if t.Name == topic {
			// This should always hit, unless kafka has a bug.
			if t.ErrorCode != 0 {
				return 0, Error(t.ErrorCode)
			}
			return len(t.Partitions), nil
		}
	}
	return 0, UnknownTopicOrPartition
}
alpergencdev added a commit to alpergencdev/kafka-go that referenced this issue Jul 9, 2024
alpergencdev added a commit to alpergencdev/kafka-go that referenced this issue Jul 9, 2024
alpergencdev added a commit to alpergencdev/kafka-go that referenced this issue Jul 13, 2024
alpergencdev added a commit to alpergencdev/kafka-go that referenced this issue Jul 13, 2024
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
Development

No branches or pull requests

1 participant