NewOffsetCommitRequest creates an OffsetCommitRequest initialized for admin use. The version-mapping logic mirrors offsetManager.constructRequest in offset_manager.go; protocol bumps must be applied to both call sites.
(conf *Config, group string)
| 88 | // The version-mapping logic mirrors offsetManager.constructRequest in |
| 89 | // offset_manager.go; protocol bumps must be applied to both call sites. |
| 90 | func NewOffsetCommitRequest(conf *Config, group string) *OffsetCommitRequest { |
| 91 | request := &OffsetCommitRequest{ |
| 92 | ConsumerGroup: group, |
| 93 | ConsumerGroupGeneration: GroupGenerationUndefined, |
| 94 | } |
| 95 | |
| 96 | if conf.Version.IsAtLeast(V2_4_0_0) { |
| 97 | // Version 8 is the first flexible version. |
| 98 | request.Version = 8 |
| 99 | } else if conf.Version.IsAtLeast(V2_3_0_0) { |
| 100 | // Version 7 adds GroupInstanceId. |
| 101 | request.Version = 7 |
| 102 | } else if conf.Version.IsAtLeast(V2_1_0_0) { |
| 103 | // Version 6 adds committed leader epoch (version 5 removes retention time). |
| 104 | request.Version = 6 |
| 105 | } else if conf.Version.IsAtLeast(V2_0_0_0) { |
| 106 | // Version 4 is the same as version 2. |
| 107 | request.Version = 4 |
| 108 | } else if conf.Version.IsAtLeast(V0_11_0_0) { |
| 109 | // Version 3 is the same as version 2. |
| 110 | request.Version = 3 |
| 111 | } else if conf.Version.IsAtLeast(V0_9_0_0) { |
| 112 | // Version 2 adds retention time and removes the commit timestamp from version 1. |
| 113 | request.Version = 2 |
| 114 | } else { |
| 115 | // Version 1 adds commit timestamp and group membership. |
| 116 | request.Version = 1 |
| 117 | } |
| 118 | |
| 119 | if request.Version >= 2 && request.Version < 5 { |
| 120 | request.RetentionTime = -1 |
| 121 | if conf.Consumer.Offsets.Retention > 0 { |
| 122 | request.RetentionTime = conf.Consumer.Offsets.Retention.Milliseconds() |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | return request |
| 127 | } |
| 128 | |
| 129 | func (r *OffsetCommitRequest) encode(pe packetEncoder) error { |
| 130 | if r.Version < 0 || r.Version > 8 { |
no test coverage detected