create_topics_request.go 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174
  1. package sarama
  2. import (
  3. "time"
  4. )
  5. type CreateTopicsRequest struct {
  6. Version int16
  7. TopicDetails map[string]*TopicDetail
  8. Timeout time.Duration
  9. ValidateOnly bool
  10. }
  11. func (c *CreateTopicsRequest) encode(pe packetEncoder) error {
  12. if err := pe.putArrayLength(len(c.TopicDetails)); err != nil {
  13. return err
  14. }
  15. for topic, detail := range c.TopicDetails {
  16. if err := pe.putString(topic); err != nil {
  17. return err
  18. }
  19. if err := detail.encode(pe); err != nil {
  20. return err
  21. }
  22. }
  23. pe.putInt32(int32(c.Timeout / time.Millisecond))
  24. if c.Version >= 1 {
  25. pe.putBool(c.ValidateOnly)
  26. }
  27. return nil
  28. }
  29. func (c *CreateTopicsRequest) decode(pd packetDecoder, version int16) (err error) {
  30. n, err := pd.getArrayLength()
  31. if err != nil {
  32. return err
  33. }
  34. c.TopicDetails = make(map[string]*TopicDetail, n)
  35. for i := 0; i < n; i++ {
  36. topic, err := pd.getString()
  37. if err != nil {
  38. return err
  39. }
  40. c.TopicDetails[topic] = new(TopicDetail)
  41. if err = c.TopicDetails[topic].decode(pd, version); err != nil {
  42. return err
  43. }
  44. }
  45. timeout, err := pd.getInt32()
  46. if err != nil {
  47. return err
  48. }
  49. c.Timeout = time.Duration(timeout) * time.Millisecond
  50. if version >= 1 {
  51. c.ValidateOnly, err = pd.getBool()
  52. if err != nil {
  53. return err
  54. }
  55. c.Version = version
  56. }
  57. return nil
  58. }
  59. func (c *CreateTopicsRequest) key() int16 {
  60. return 19
  61. }
  62. func (c *CreateTopicsRequest) version() int16 {
  63. return c.Version
  64. }
  65. func (c *CreateTopicsRequest) requiredVersion() KafkaVersion {
  66. switch c.Version {
  67. case 2:
  68. return V1_0_0_0
  69. case 1:
  70. return V0_11_0_0
  71. default:
  72. return V0_10_1_0
  73. }
  74. }
  75. type TopicDetail struct {
  76. NumPartitions int32
  77. ReplicationFactor int16
  78. ReplicaAssignment map[int32][]int32
  79. ConfigEntries map[string]*string
  80. }
  81. func (t *TopicDetail) encode(pe packetEncoder) error {
  82. pe.putInt32(t.NumPartitions)
  83. pe.putInt16(t.ReplicationFactor)
  84. if err := pe.putArrayLength(len(t.ReplicaAssignment)); err != nil {
  85. return err
  86. }
  87. for partition, assignment := range t.ReplicaAssignment {
  88. pe.putInt32(partition)
  89. if err := pe.putInt32Array(assignment); err != nil {
  90. return err
  91. }
  92. }
  93. if err := pe.putArrayLength(len(t.ConfigEntries)); err != nil {
  94. return err
  95. }
  96. for configKey, configValue := range t.ConfigEntries {
  97. if err := pe.putString(configKey); err != nil {
  98. return err
  99. }
  100. if err := pe.putNullableString(configValue); err != nil {
  101. return err
  102. }
  103. }
  104. return nil
  105. }
  106. func (t *TopicDetail) decode(pd packetDecoder, version int16) (err error) {
  107. if t.NumPartitions, err = pd.getInt32(); err != nil {
  108. return err
  109. }
  110. if t.ReplicationFactor, err = pd.getInt16(); err != nil {
  111. return err
  112. }
  113. n, err := pd.getArrayLength()
  114. if err != nil {
  115. return err
  116. }
  117. if n > 0 {
  118. t.ReplicaAssignment = make(map[int32][]int32, n)
  119. for i := 0; i < n; i++ {
  120. replica, err := pd.getInt32()
  121. if err != nil {
  122. return err
  123. }
  124. if t.ReplicaAssignment[replica], err = pd.getInt32Array(); err != nil {
  125. return err
  126. }
  127. }
  128. }
  129. n, err = pd.getArrayLength()
  130. if err != nil {
  131. return err
  132. }
  133. if n > 0 {
  134. t.ConfigEntries = make(map[string]*string, n)
  135. for i := 0; i < n; i++ {
  136. configKey, err := pd.getString()
  137. if err != nil {
  138. return err
  139. }
  140. if t.ConfigEntries[configKey], err = pd.getNullableString(); err != nil {
  141. return err
  142. }
  143. }
  144. }
  145. return nil
  146. }