describe_configs_response.go 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319
  1. package sarama
  2. import (
  3. "fmt"
  4. "time"
  5. )
  6. type ConfigSource int8
  7. func (s ConfigSource) String() string {
  8. switch s {
  9. case SourceUnknown:
  10. return "Unknown"
  11. case SourceTopic:
  12. return "Topic"
  13. case SourceDynamicBroker:
  14. return "DynamicBroker"
  15. case SourceDynamicDefaultBroker:
  16. return "DynamicDefaultBroker"
  17. case SourceStaticBroker:
  18. return "StaticBroker"
  19. case SourceDefault:
  20. return "Default"
  21. }
  22. return fmt.Sprintf("Source Invalid: %d", int(s))
  23. }
  24. const (
  25. SourceUnknown ConfigSource = iota
  26. SourceTopic
  27. SourceDynamicBroker
  28. SourceDynamicDefaultBroker
  29. SourceStaticBroker
  30. SourceDefault
  31. )
  32. type DescribeConfigsResponse struct {
  33. Version int16
  34. ThrottleTime time.Duration
  35. Resources []*ResourceResponse
  36. }
  37. type ResourceResponse struct {
  38. ErrorCode int16
  39. ErrorMsg string
  40. Type ConfigResourceType
  41. Name string
  42. Configs []*ConfigEntry
  43. }
  44. type ConfigEntry struct {
  45. Name string
  46. Value string
  47. ReadOnly bool
  48. Default bool
  49. Source ConfigSource
  50. Sensitive bool
  51. Synonyms []*ConfigSynonym
  52. }
  53. type ConfigSynonym struct {
  54. ConfigName string
  55. ConfigValue string
  56. Source ConfigSource
  57. }
  58. func (r *DescribeConfigsResponse) encode(pe packetEncoder) (err error) {
  59. pe.putInt32(int32(r.ThrottleTime / time.Millisecond))
  60. if err = pe.putArrayLength(len(r.Resources)); err != nil {
  61. return err
  62. }
  63. for _, c := range r.Resources {
  64. if err = c.encode(pe, r.Version); err != nil {
  65. return err
  66. }
  67. }
  68. return nil
  69. }
  70. func (r *DescribeConfigsResponse) decode(pd packetDecoder, version int16) (err error) {
  71. r.Version = version
  72. throttleTime, err := pd.getInt32()
  73. if err != nil {
  74. return err
  75. }
  76. r.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
  77. n, err := pd.getArrayLength()
  78. if err != nil {
  79. return err
  80. }
  81. r.Resources = make([]*ResourceResponse, n)
  82. for i := 0; i < n; i++ {
  83. rr := &ResourceResponse{}
  84. if err := rr.decode(pd, version); err != nil {
  85. return err
  86. }
  87. r.Resources[i] = rr
  88. }
  89. return nil
  90. }
  91. func (r *DescribeConfigsResponse) key() int16 {
  92. return 32
  93. }
  94. func (r *DescribeConfigsResponse) version() int16 {
  95. return r.Version
  96. }
  97. func (r *DescribeConfigsResponse) requiredVersion() KafkaVersion {
  98. switch r.Version {
  99. case 1:
  100. return V1_0_0_0
  101. case 2:
  102. return V2_0_0_0
  103. default:
  104. return V0_11_0_0
  105. }
  106. }
  107. func (r *ResourceResponse) encode(pe packetEncoder, version int16) (err error) {
  108. pe.putInt16(r.ErrorCode)
  109. if err = pe.putString(r.ErrorMsg); err != nil {
  110. return err
  111. }
  112. pe.putInt8(int8(r.Type))
  113. if err = pe.putString(r.Name); err != nil {
  114. return err
  115. }
  116. if err = pe.putArrayLength(len(r.Configs)); err != nil {
  117. return err
  118. }
  119. for _, c := range r.Configs {
  120. if err = c.encode(pe, version); err != nil {
  121. return err
  122. }
  123. }
  124. return nil
  125. }
  126. func (r *ResourceResponse) decode(pd packetDecoder, version int16) (err error) {
  127. ec, err := pd.getInt16()
  128. if err != nil {
  129. return err
  130. }
  131. r.ErrorCode = ec
  132. em, err := pd.getString()
  133. if err != nil {
  134. return err
  135. }
  136. r.ErrorMsg = em
  137. t, err := pd.getInt8()
  138. if err != nil {
  139. return err
  140. }
  141. r.Type = ConfigResourceType(t)
  142. name, err := pd.getString()
  143. if err != nil {
  144. return err
  145. }
  146. r.Name = name
  147. n, err := pd.getArrayLength()
  148. if err != nil {
  149. return err
  150. }
  151. r.Configs = make([]*ConfigEntry, n)
  152. for i := 0; i < n; i++ {
  153. c := &ConfigEntry{}
  154. if err := c.decode(pd, version); err != nil {
  155. return err
  156. }
  157. r.Configs[i] = c
  158. }
  159. return nil
  160. }
  161. func (r *ConfigEntry) encode(pe packetEncoder, version int16) (err error) {
  162. if err = pe.putString(r.Name); err != nil {
  163. return err
  164. }
  165. if err = pe.putString(r.Value); err != nil {
  166. return err
  167. }
  168. pe.putBool(r.ReadOnly)
  169. if version <= 0 {
  170. pe.putBool(r.Default)
  171. pe.putBool(r.Sensitive)
  172. } else {
  173. pe.putInt8(int8(r.Source))
  174. pe.putBool(r.Sensitive)
  175. if err := pe.putArrayLength(len(r.Synonyms)); err != nil {
  176. return err
  177. }
  178. for _, c := range r.Synonyms {
  179. if err = c.encode(pe, version); err != nil {
  180. return err
  181. }
  182. }
  183. }
  184. return nil
  185. }
  186. //https://cwiki.apache.org/confluence/display/KAFKA/KIP-226+-+Dynamic+Broker+Configuration
  187. func (r *ConfigEntry) decode(pd packetDecoder, version int16) (err error) {
  188. if version == 0 {
  189. r.Source = SourceUnknown
  190. }
  191. name, err := pd.getString()
  192. if err != nil {
  193. return err
  194. }
  195. r.Name = name
  196. value, err := pd.getString()
  197. if err != nil {
  198. return err
  199. }
  200. r.Value = value
  201. read, err := pd.getBool()
  202. if err != nil {
  203. return err
  204. }
  205. r.ReadOnly = read
  206. if version == 0 {
  207. defaultB, err := pd.getBool()
  208. if err != nil {
  209. return err
  210. }
  211. r.Default = defaultB
  212. } else {
  213. source, err := pd.getInt8()
  214. if err != nil {
  215. return err
  216. }
  217. r.Source = ConfigSource(source)
  218. }
  219. sensitive, err := pd.getBool()
  220. if err != nil {
  221. return err
  222. }
  223. r.Sensitive = sensitive
  224. if version > 0 {
  225. n, err := pd.getArrayLength()
  226. if err != nil {
  227. return err
  228. }
  229. r.Synonyms = make([]*ConfigSynonym, n)
  230. for i := 0; i < n; i++ {
  231. s := &ConfigSynonym{}
  232. if err := s.decode(pd, version); err != nil {
  233. return err
  234. }
  235. r.Synonyms[i] = s
  236. }
  237. }
  238. return nil
  239. }
  240. func (c *ConfigSynonym) encode(pe packetEncoder, version int16) (err error) {
  241. err = pe.putString(c.ConfigName)
  242. if err != nil {
  243. return err
  244. }
  245. err = pe.putString(c.ConfigValue)
  246. if err != nil {
  247. return err
  248. }
  249. pe.putInt8(int8(c.Source))
  250. return nil
  251. }
  252. func (c *ConfigSynonym) decode(pd packetDecoder, version int16) error {
  253. name, err := pd.getString()
  254. if err != nil {
  255. return nil
  256. }
  257. c.ConfigName = name
  258. value, err := pd.getString()
  259. if err != nil {
  260. return nil
  261. }
  262. c.ConfigValue = value
  263. source, err := pd.getInt8()
  264. if err != nil {
  265. return nil
  266. }
  267. c.Source = ConfigSource(source)
  268. return nil
  269. }