delete_records_request.go 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126
  1. package sarama
  2. import (
  3. "sort"
  4. "time"
  5. )
  6. // request message format is:
  7. // [topic] timeout(int32)
  8. // where topic is:
  9. // name(string) [partition]
  10. // where partition is:
  11. // id(int32) offset(int64)
  12. type DeleteRecordsRequest struct {
  13. Topics map[string]*DeleteRecordsRequestTopic
  14. Timeout time.Duration
  15. }
  16. func (d *DeleteRecordsRequest) encode(pe packetEncoder) error {
  17. if err := pe.putArrayLength(len(d.Topics)); err != nil {
  18. return err
  19. }
  20. keys := make([]string, 0, len(d.Topics))
  21. for topic := range d.Topics {
  22. keys = append(keys, topic)
  23. }
  24. sort.Strings(keys)
  25. for _, topic := range keys {
  26. if err := pe.putString(topic); err != nil {
  27. return err
  28. }
  29. if err := d.Topics[topic].encode(pe); err != nil {
  30. return err
  31. }
  32. }
  33. pe.putInt32(int32(d.Timeout / time.Millisecond))
  34. return nil
  35. }
  36. func (d *DeleteRecordsRequest) decode(pd packetDecoder, version int16) error {
  37. n, err := pd.getArrayLength()
  38. if err != nil {
  39. return err
  40. }
  41. if n > 0 {
  42. d.Topics = make(map[string]*DeleteRecordsRequestTopic, n)
  43. for i := 0; i < n; i++ {
  44. topic, err := pd.getString()
  45. if err != nil {
  46. return err
  47. }
  48. details := new(DeleteRecordsRequestTopic)
  49. if err = details.decode(pd, version); err != nil {
  50. return err
  51. }
  52. d.Topics[topic] = details
  53. }
  54. }
  55. timeout, err := pd.getInt32()
  56. if err != nil {
  57. return err
  58. }
  59. d.Timeout = time.Duration(timeout) * time.Millisecond
  60. return nil
  61. }
  62. func (d *DeleteRecordsRequest) key() int16 {
  63. return 21
  64. }
  65. func (d *DeleteRecordsRequest) version() int16 {
  66. return 0
  67. }
  68. func (d *DeleteRecordsRequest) requiredVersion() KafkaVersion {
  69. return V0_11_0_0
  70. }
  71. type DeleteRecordsRequestTopic struct {
  72. PartitionOffsets map[int32]int64 // partition => offset
  73. }
  74. func (t *DeleteRecordsRequestTopic) encode(pe packetEncoder) error {
  75. if err := pe.putArrayLength(len(t.PartitionOffsets)); err != nil {
  76. return err
  77. }
  78. keys := make([]int32, 0, len(t.PartitionOffsets))
  79. for partition := range t.PartitionOffsets {
  80. keys = append(keys, partition)
  81. }
  82. sort.Slice(keys, func(i, j int) bool { return keys[i] < keys[j] })
  83. for _, partition := range keys {
  84. pe.putInt32(partition)
  85. pe.putInt64(t.PartitionOffsets[partition])
  86. }
  87. return nil
  88. }
  89. func (t *DeleteRecordsRequestTopic) decode(pd packetDecoder, version int16) error {
  90. n, err := pd.getArrayLength()
  91. if err != nil {
  92. return err
  93. }
  94. if n > 0 {
  95. t.PartitionOffsets = make(map[int32]int64, n)
  96. for i := 0; i < n; i++ {
  97. partition, err := pd.getInt32()
  98. if err != nil {
  99. return err
  100. }
  101. offset, err := pd.getInt64()
  102. if err != nil {
  103. return err
  104. }
  105. t.PartitionOffsets[partition] = offset
  106. }
  107. }
  108. return nil
  109. }