create_partitions_response.go 2.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129
  1. package sarama
  2. import (
  3. "fmt"
  4. "time"
  5. )
  6. type CreatePartitionsResponse struct {
  7. Version int16
  8. ThrottleTime time.Duration
  9. TopicPartitionErrors map[string]*TopicPartitionError
  10. }
  11. func (c *CreatePartitionsResponse) encode(pe packetEncoder) error {
  12. pe.putInt32(int32(c.ThrottleTime / time.Millisecond))
  13. if err := pe.putArrayLength(len(c.TopicPartitionErrors)); err != nil {
  14. return err
  15. }
  16. for topic, partitionError := range c.TopicPartitionErrors {
  17. if err := pe.putString(topic); err != nil {
  18. return err
  19. }
  20. if err := partitionError.encode(pe); err != nil {
  21. return err
  22. }
  23. }
  24. return nil
  25. }
  26. func (c *CreatePartitionsResponse) decode(pd packetDecoder, version int16) (err error) {
  27. throttleTime, err := pd.getInt32()
  28. if err != nil {
  29. return err
  30. }
  31. c.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
  32. n, err := pd.getArrayLength()
  33. if err != nil {
  34. return err
  35. }
  36. c.TopicPartitionErrors = make(map[string]*TopicPartitionError, n)
  37. for i := 0; i < n; i++ {
  38. topic, err := pd.getString()
  39. if err != nil {
  40. return err
  41. }
  42. c.TopicPartitionErrors[topic] = new(TopicPartitionError)
  43. if err := c.TopicPartitionErrors[topic].decode(pd, version); err != nil {
  44. return err
  45. }
  46. }
  47. return nil
  48. }
  49. func (r *CreatePartitionsResponse) key() int16 {
  50. return 37
  51. }
  52. func (r *CreatePartitionsResponse) version() int16 {
  53. return r.Version
  54. }
  55. func (r *CreatePartitionsResponse) headerVersion() int16 {
  56. return 0
  57. }
  58. func (r *CreatePartitionsResponse) isValidVersion() bool {
  59. return r.Version >= 0 && r.Version <= 1
  60. }
  61. func (r *CreatePartitionsResponse) requiredVersion() KafkaVersion {
  62. switch r.Version {
  63. case 1:
  64. return V2_0_0_0
  65. case 0:
  66. return V1_0_0_0
  67. default:
  68. return V2_0_0_0
  69. }
  70. }
  71. func (r *CreatePartitionsResponse) throttleTime() time.Duration {
  72. return r.ThrottleTime
  73. }
  74. type TopicPartitionError struct {
  75. Err KError
  76. ErrMsg *string
  77. }
  78. func (t *TopicPartitionError) Error() string {
  79. text := t.Err.Error()
  80. if t.ErrMsg != nil {
  81. text = fmt.Sprintf("%s - %s", text, *t.ErrMsg)
  82. }
  83. return text
  84. }
  85. func (t *TopicPartitionError) Unwrap() error {
  86. return t.Err
  87. }
  88. func (t *TopicPartitionError) encode(pe packetEncoder) error {
  89. pe.putInt16(int16(t.Err))
  90. if err := pe.putNullableString(t.ErrMsg); err != nil {
  91. return err
  92. }
  93. return nil
  94. }
  95. func (t *TopicPartitionError) decode(pd packetDecoder, version int16) (err error) {
  96. kerr, err := pd.getInt16()
  97. if err != nil {
  98. return err
  99. }
  100. t.Err = KError(kerr)
  101. if t.ErrMsg, err = pd.getNullableString(); err != nil {
  102. return err
  103. }
  104. return nil
  105. }