| 332 | } |
| 333 | |
| 334 | func (om *offsetManager) constructRequestFor(targets partitionTargets) *OffsetCommitRequest { |
| 335 | r := &OffsetCommitRequest{ |
| 336 | Version: 1, |
| 337 | ConsumerGroup: om.group, |
| 338 | ConsumerID: om.memberID, |
| 339 | // om.generation is read under generationLock, held by flushToBrokerFor |
| 340 | ConsumerGroupGeneration: om.generation, |
| 341 | } |
| 342 | // Version 1 adds timestamp and group membership information, as well as the commit timestamp. |
| 343 | // |
| 344 | // Version 2 adds retention time. It removes the commit timestamp added in version 1. |
| 345 | if om.conf.Version.IsAtLeast(V0_9_0_0) { |
| 346 | r.Version = 2 |
| 347 | } |
| 348 | // Version 3 and 4 are the same as version 2. |
| 349 | if om.conf.Version.IsAtLeast(V0_11_0_0) { |
| 350 | r.Version = 3 |
| 351 | } |
| 352 | if om.conf.Version.IsAtLeast(V2_0_0_0) { |
| 353 | r.Version = 4 |
| 354 | } |
| 355 | // Version 5 removes the retention time, which is now controlled only by a broker configuration. |
| 356 | // |
| 357 | // Version 6 adds the leader epoch for fencing. |
| 358 | if om.conf.Version.IsAtLeast(V2_1_0_0) { |
| 359 | r.Version = 6 |
| 360 | } |
| 361 | // version 7 adds a new field called groupInstanceId to indicate member identity across restarts. |
| 362 | if om.conf.Version.IsAtLeast(V2_3_0_0) { |
| 363 | r.Version = 7 |
| 364 | r.GroupInstanceId = om.groupInstanceId |
| 365 | } |
| 366 | // Version 8 is the first flexible version. |
| 367 | if om.conf.Version.IsAtLeast(V2_4_0_0) { |
| 368 | r.Version = 8 |
| 369 | } |
| 370 | |
| 371 | // commit timestamp was only briefly supported in V1 where we set it to |
| 372 | // ReceiveTime (-1) to tell the broker to set it to the time when the commit |
| 373 | // request was received |
| 374 | var commitTimestamp int64 |
| 375 | if r.Version == 1 { |
| 376 | commitTimestamp = ReceiveTime |
| 377 | } |
| 378 | |
| 379 | // request controlled retention was only supported from V2-V4 (it became |
| 380 | // broker-only after that) so if the user has set the config options then |
| 381 | // flow those through as retention time on the commit request. |
| 382 | if r.Version >= 2 && r.Version < 5 { |
| 383 | // Map Sarama's default of 0 to Kafka's default of -1 |
| 384 | r.RetentionTime = -1 |
| 385 | if om.conf.Consumer.Offsets.Retention > 0 { |
| 386 | r.RetentionTime = int64(om.conf.Consumer.Offsets.Retention / time.Millisecond) |
| 387 | } |
| 388 | } |
| 389 | |
| 390 | om.pomsLock.RLock() |
| 391 | defer om.pomsLock.RUnlock() |