incremental_alter_configs_request.go 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178
  1. package sarama
  2. type IncrementalAlterConfigsOperation int8
  3. const (
  4. IncrementalAlterConfigsOperationSet IncrementalAlterConfigsOperation = iota
  5. IncrementalAlterConfigsOperationDelete
  6. IncrementalAlterConfigsOperationAppend
  7. IncrementalAlterConfigsOperationSubtract
  8. )
  9. // IncrementalAlterConfigsRequest is an incremental alter config request type
  10. type IncrementalAlterConfigsRequest struct {
  11. Version int16
  12. Resources []*IncrementalAlterConfigsResource
  13. ValidateOnly bool
  14. }
  15. type IncrementalAlterConfigsResource struct {
  16. Type ConfigResourceType
  17. Name string
  18. ConfigEntries map[string]IncrementalAlterConfigsEntry
  19. }
  20. type IncrementalAlterConfigsEntry struct {
  21. Operation IncrementalAlterConfigsOperation
  22. Value *string
  23. }
  24. func (a *IncrementalAlterConfigsRequest) encode(pe packetEncoder) error {
  25. if err := pe.putArrayLength(len(a.Resources)); err != nil {
  26. return err
  27. }
  28. for _, r := range a.Resources {
  29. if err := r.encode(pe); err != nil {
  30. return err
  31. }
  32. }
  33. pe.putBool(a.ValidateOnly)
  34. return nil
  35. }
  36. func (a *IncrementalAlterConfigsRequest) decode(pd packetDecoder, version int16) error {
  37. resourceCount, err := pd.getArrayLength()
  38. if err != nil {
  39. return err
  40. }
  41. a.Resources = make([]*IncrementalAlterConfigsResource, resourceCount)
  42. for i := range a.Resources {
  43. r := &IncrementalAlterConfigsResource{}
  44. err = r.decode(pd, version)
  45. if err != nil {
  46. return err
  47. }
  48. a.Resources[i] = r
  49. }
  50. validateOnly, err := pd.getBool()
  51. if err != nil {
  52. return err
  53. }
  54. a.ValidateOnly = validateOnly
  55. return nil
  56. }
  57. func (a *IncrementalAlterConfigsResource) encode(pe packetEncoder) error {
  58. pe.putInt8(int8(a.Type))
  59. if err := pe.putString(a.Name); err != nil {
  60. return err
  61. }
  62. if err := pe.putArrayLength(len(a.ConfigEntries)); err != nil {
  63. return err
  64. }
  65. for name, e := range a.ConfigEntries {
  66. if err := pe.putString(name); err != nil {
  67. return err
  68. }
  69. if err := e.encode(pe); err != nil {
  70. return err
  71. }
  72. }
  73. return nil
  74. }
  75. func (a *IncrementalAlterConfigsResource) decode(pd packetDecoder, version int16) error {
  76. t, err := pd.getInt8()
  77. if err != nil {
  78. return err
  79. }
  80. a.Type = ConfigResourceType(t)
  81. name, err := pd.getString()
  82. if err != nil {
  83. return err
  84. }
  85. a.Name = name
  86. n, err := pd.getArrayLength()
  87. if err != nil {
  88. return err
  89. }
  90. if n > 0 {
  91. a.ConfigEntries = make(map[string]IncrementalAlterConfigsEntry, n)
  92. for i := 0; i < n; i++ {
  93. name, err := pd.getString()
  94. if err != nil {
  95. return err
  96. }
  97. var v IncrementalAlterConfigsEntry
  98. if err := v.decode(pd, version); err != nil {
  99. return err
  100. }
  101. a.ConfigEntries[name] = v
  102. }
  103. }
  104. return err
  105. }
  106. func (a *IncrementalAlterConfigsEntry) encode(pe packetEncoder) error {
  107. pe.putInt8(int8(a.Operation))
  108. if err := pe.putNullableString(a.Value); err != nil {
  109. return err
  110. }
  111. return nil
  112. }
  113. func (a *IncrementalAlterConfigsEntry) decode(pd packetDecoder, version int16) error {
  114. t, err := pd.getInt8()
  115. if err != nil {
  116. return err
  117. }
  118. a.Operation = IncrementalAlterConfigsOperation(t)
  119. s, err := pd.getNullableString()
  120. if err != nil {
  121. return err
  122. }
  123. a.Value = s
  124. return nil
  125. }
  126. func (a *IncrementalAlterConfigsRequest) key() int16 {
  127. return 44
  128. }
  129. func (a *IncrementalAlterConfigsRequest) version() int16 {
  130. return a.Version
  131. }
  132. func (a *IncrementalAlterConfigsRequest) headerVersion() int16 {
  133. return 1
  134. }
  135. func (a *IncrementalAlterConfigsRequest) isValidVersion() bool {
  136. return a.Version == 0
  137. }
  138. func (a *IncrementalAlterConfigsRequest) requiredVersion() KafkaVersion {
  139. return V2_3_0_0
  140. }