describe_configs_response.go 6.1 KB

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