offset_request.go 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156
  1. package sarama
  2. type offsetRequestBlock struct {
  3. time int64
  4. maxOffsets int32 // Only used in version 0
  5. }
  6. func (b *offsetRequestBlock) encode(pe packetEncoder, version int16) error {
  7. pe.putInt64(int64(b.time))
  8. if version == 0 {
  9. pe.putInt32(b.maxOffsets)
  10. }
  11. return nil
  12. }
  13. func (b *offsetRequestBlock) decode(pd packetDecoder, version int16) (err error) {
  14. if b.time, err = pd.getInt64(); err != nil {
  15. return err
  16. }
  17. if version == 0 {
  18. if b.maxOffsets, err = pd.getInt32(); err != nil {
  19. return err
  20. }
  21. }
  22. return nil
  23. }
  24. type OffsetRequest struct {
  25. Version int16
  26. replicaID int32
  27. isReplicaIDSet bool
  28. blocks map[string]map[int32]*offsetRequestBlock
  29. }
  30. func (r *OffsetRequest) encode(pe packetEncoder) error {
  31. if r.isReplicaIDSet {
  32. pe.putInt32(r.replicaID)
  33. } else {
  34. // default replica ID is always -1 for clients
  35. pe.putInt32(-1)
  36. }
  37. err := pe.putArrayLength(len(r.blocks))
  38. if err != nil {
  39. return err
  40. }
  41. for topic, partitions := range r.blocks {
  42. err = pe.putString(topic)
  43. if err != nil {
  44. return err
  45. }
  46. err = pe.putArrayLength(len(partitions))
  47. if err != nil {
  48. return err
  49. }
  50. for partition, block := range partitions {
  51. pe.putInt32(partition)
  52. if err = block.encode(pe, r.Version); err != nil {
  53. return err
  54. }
  55. }
  56. }
  57. return nil
  58. }
  59. func (r *OffsetRequest) decode(pd packetDecoder, version int16) error {
  60. r.Version = version
  61. replicaID, err := pd.getInt32()
  62. if err != nil {
  63. return err
  64. }
  65. if replicaID >= 0 {
  66. r.SetReplicaID(replicaID)
  67. }
  68. blockCount, err := pd.getArrayLength()
  69. if err != nil {
  70. return err
  71. }
  72. if blockCount == 0 {
  73. return nil
  74. }
  75. r.blocks = make(map[string]map[int32]*offsetRequestBlock)
  76. for i := 0; i < blockCount; i++ {
  77. topic, err := pd.getString()
  78. if err != nil {
  79. return err
  80. }
  81. partitionCount, err := pd.getArrayLength()
  82. if err != nil {
  83. return err
  84. }
  85. r.blocks[topic] = make(map[int32]*offsetRequestBlock)
  86. for j := 0; j < partitionCount; j++ {
  87. partition, err := pd.getInt32()
  88. if err != nil {
  89. return err
  90. }
  91. block := &offsetRequestBlock{}
  92. if err := block.decode(pd, version); err != nil {
  93. return err
  94. }
  95. r.blocks[topic][partition] = block
  96. }
  97. }
  98. return nil
  99. }
  100. func (r *OffsetRequest) key() int16 {
  101. return 2
  102. }
  103. func (r *OffsetRequest) version() int16 {
  104. return r.Version
  105. }
  106. func (r *OffsetRequest) requiredVersion() KafkaVersion {
  107. switch r.Version {
  108. case 1:
  109. return V0_10_1_0
  110. default:
  111. return MinVersion
  112. }
  113. }
  114. func (r *OffsetRequest) SetReplicaID(id int32) {
  115. r.replicaID = id
  116. r.isReplicaIDSet = true
  117. }
  118. func (r *OffsetRequest) ReplicaID() int32 {
  119. if r.isReplicaIDSet {
  120. return r.replicaID
  121. }
  122. return -1
  123. }
  124. func (r *OffsetRequest) AddBlock(topic string, partitionID int32, time int64, maxOffsets int32) {
  125. if r.blocks == nil {
  126. r.blocks = make(map[string]map[int32]*offsetRequestBlock)
  127. }
  128. if r.blocks[topic] == nil {
  129. r.blocks[topic] = make(map[int32]*offsetRequestBlock)
  130. }
  131. tmp := new(offsetRequestBlock)
  132. tmp.time = time
  133. if r.Version == 0 {
  134. tmp.maxOffsets = maxOffsets
  135. }
  136. r.blocks[topic][partitionID] = tmp
  137. }