add_partitions_to_txn_response.go 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. package sarama
  2. import (
  3. "time"
  4. )
  5. //AddPartitionsToTxnResponse is a partition errors to transaction type
  6. type AddPartitionsToTxnResponse struct {
  7. ThrottleTime time.Duration
  8. Errors map[string][]*PartitionError
  9. }
  10. func (a *AddPartitionsToTxnResponse) encode(pe packetEncoder) error {
  11. pe.putInt32(int32(a.ThrottleTime / time.Millisecond))
  12. if err := pe.putArrayLength(len(a.Errors)); err != nil {
  13. return err
  14. }
  15. for topic, e := range a.Errors {
  16. if err := pe.putString(topic); err != nil {
  17. return err
  18. }
  19. if err := pe.putArrayLength(len(e)); err != nil {
  20. return err
  21. }
  22. for _, partitionError := range e {
  23. if err := partitionError.encode(pe); err != nil {
  24. return err
  25. }
  26. }
  27. }
  28. return nil
  29. }
  30. func (a *AddPartitionsToTxnResponse) decode(pd packetDecoder, version int16) (err error) {
  31. throttleTime, err := pd.getInt32()
  32. if err != nil {
  33. return err
  34. }
  35. a.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
  36. n, err := pd.getArrayLength()
  37. if err != nil {
  38. return err
  39. }
  40. a.Errors = make(map[string][]*PartitionError)
  41. for i := 0; i < n; i++ {
  42. topic, err := pd.getString()
  43. if err != nil {
  44. return err
  45. }
  46. m, err := pd.getArrayLength()
  47. if err != nil {
  48. return err
  49. }
  50. a.Errors[topic] = make([]*PartitionError, m)
  51. for j := 0; j < m; j++ {
  52. a.Errors[topic][j] = new(PartitionError)
  53. if err := a.Errors[topic][j].decode(pd, version); err != nil {
  54. return err
  55. }
  56. }
  57. }
  58. return nil
  59. }
  60. func (a *AddPartitionsToTxnResponse) key() int16 {
  61. return 24
  62. }
  63. func (a *AddPartitionsToTxnResponse) version() int16 {
  64. return 0
  65. }
  66. func (a *AddPartitionsToTxnResponse) requiredVersion() KafkaVersion {
  67. return V0_11_0_0
  68. }
  69. //PartitionError is a partition error type
  70. type PartitionError struct {
  71. Partition int32
  72. Err KError
  73. }
  74. func (p *PartitionError) encode(pe packetEncoder) error {
  75. pe.putInt32(p.Partition)
  76. pe.putInt16(int16(p.Err))
  77. return nil
  78. }
  79. func (p *PartitionError) decode(pd packetDecoder, version int16) (err error) {
  80. if p.Partition, err = pd.getInt32(); err != nil {
  81. return err
  82. }
  83. kerr, err := pd.getInt16()
  84. if err != nil {
  85. return err
  86. }
  87. p.Err = KError(kerr)
  88. return nil
  89. }