createtopics.go 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374
  1. package createtopics
  2. import "github.com/segmentio/kafka-go/protocol"
  3. func init() {
  4. protocol.Register(&Request{}, &Response{})
  5. }
  6. type Request struct {
  7. // We need at least one tagged field to indicate that v5+ uses "flexible"
  8. // messages.
  9. _ struct{} `kafka:"min=v5,max=v5,tag"`
  10. Topics []RequestTopic `kafka:"min=v0,max=v5"`
  11. TimeoutMs int32 `kafka:"min=v0,max=v5"`
  12. ValidateOnly bool `kafka:"min=v1,max=v5"`
  13. }
  14. func (r *Request) ApiKey() protocol.ApiKey { return protocol.CreateTopics }
  15. func (r *Request) Broker(cluster protocol.Cluster) (protocol.Broker, error) {
  16. return cluster.Brokers[cluster.Controller], nil
  17. }
  18. type RequestTopic struct {
  19. Name string `kafka:"min=v0,max=v5"`
  20. NumPartitions int32 `kafka:"min=v0,max=v5"`
  21. ReplicationFactor int16 `kafka:"min=v0,max=v5"`
  22. Assignments []RequestAssignment `kafka:"min=v0,max=v5"`
  23. Configs []RequestConfig `kafka:"min=v0,max=v5"`
  24. }
  25. type RequestAssignment struct {
  26. PartitionIndex int32 `kafka:"min=v0,max=v5"`
  27. BrokerIDs []int32 `kafka:"min=v0,max=v5"`
  28. }
  29. type RequestConfig struct {
  30. Name string `kafka:"min=v0,max=v5"`
  31. Value string `kafka:"min=v0,max=v5,nullable"`
  32. }
  33. type Response struct {
  34. // We need at least one tagged field to indicate that v5+ uses "flexible"
  35. // messages.
  36. _ struct{} `kafka:"min=v5,max=v5,tag"`
  37. ThrottleTimeMs int32 `kafka:"min=v2,max=v5"`
  38. Topics []ResponseTopic `kafka:"min=v0,max=v5"`
  39. }
  40. func (r *Response) ApiKey() protocol.ApiKey { return protocol.CreateTopics }
  41. type ResponseTopic struct {
  42. Name string `kafka:"min=v0,max=v5"`
  43. ErrorCode int16 `kafka:"min=v0,max=v5"`
  44. ErrorMessage string `kafka:"min=v1,max=v5,nullable"`
  45. NumPartitions int32 `kafka:"min=v5,max=v5"`
  46. ReplicationFactor int16 `kafka:"min=v5,max=v5"`
  47. Configs []ResponseTopicConfig `kafka:"min=v5,max=v5"`
  48. }
  49. type ResponseTopicConfig struct {
  50. Name string `kafka:"min=v5,max=v5"`
  51. Value string `kafka:"min=v5,max=v5,nullable"`
  52. ReadOnly bool `kafka:"min=v5,max=v5"`
  53. ConfigSource int8 `kafka:"min=v5,max=v5"`
  54. IsSensitive bool `kafka:"min=v5,max=v5"`
  55. }
  56. var (
  57. _ protocol.BrokerMessage = (*Request)(nil)
  58. )