offset_commit_response.go 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. package sarama
  2. type OffsetCommitResponse struct {
  3. Version int16
  4. ThrottleTimeMs int32
  5. Errors map[string]map[int32]KError
  6. }
  7. func (r *OffsetCommitResponse) AddError(topic string, partition int32, kerror KError) {
  8. if r.Errors == nil {
  9. r.Errors = make(map[string]map[int32]KError)
  10. }
  11. partitions := r.Errors[topic]
  12. if partitions == nil {
  13. partitions = make(map[int32]KError)
  14. r.Errors[topic] = partitions
  15. }
  16. partitions[partition] = kerror
  17. }
  18. func (r *OffsetCommitResponse) encode(pe packetEncoder) error {
  19. if r.Version >= 3 {
  20. pe.putInt32(r.ThrottleTimeMs)
  21. }
  22. if err := pe.putArrayLength(len(r.Errors)); err != nil {
  23. return err
  24. }
  25. for topic, partitions := range r.Errors {
  26. if err := pe.putString(topic); err != nil {
  27. return err
  28. }
  29. if err := pe.putArrayLength(len(partitions)); err != nil {
  30. return err
  31. }
  32. for partition, kerror := range partitions {
  33. pe.putInt32(partition)
  34. pe.putInt16(int16(kerror))
  35. }
  36. }
  37. return nil
  38. }
  39. func (r *OffsetCommitResponse) decode(pd packetDecoder, version int16) (err error) {
  40. r.Version = version
  41. if version >= 3 {
  42. r.ThrottleTimeMs, err = pd.getInt32()
  43. if err != nil {
  44. return err
  45. }
  46. }
  47. numTopics, err := pd.getArrayLength()
  48. if err != nil || numTopics == 0 {
  49. return err
  50. }
  51. r.Errors = make(map[string]map[int32]KError, numTopics)
  52. for i := 0; i < numTopics; i++ {
  53. name, err := pd.getString()
  54. if err != nil {
  55. return err
  56. }
  57. numErrors, err := pd.getArrayLength()
  58. if err != nil {
  59. return err
  60. }
  61. r.Errors[name] = make(map[int32]KError, numErrors)
  62. for j := 0; j < numErrors; j++ {
  63. id, err := pd.getInt32()
  64. if err != nil {
  65. return err
  66. }
  67. tmp, err := pd.getInt16()
  68. if err != nil {
  69. return err
  70. }
  71. r.Errors[name][id] = KError(tmp)
  72. }
  73. }
  74. return nil
  75. }
  76. func (r *OffsetCommitResponse) key() int16 {
  77. return 8
  78. }
  79. func (r *OffsetCommitResponse) version() int16 {
  80. return r.Version
  81. }
  82. func (r *OffsetCommitResponse) requiredVersion() KafkaVersion {
  83. switch r.Version {
  84. case 1:
  85. return V0_8_2_0
  86. case 2:
  87. return V0_9_0_0
  88. case 3:
  89. return V0_11_0_0
  90. case 4:
  91. return V2_0_0_0
  92. default:
  93. return MinVersion
  94. }
  95. }