Skip to content

Commit

Permalink
Merge pull request #1056 from kenschneider18/document-recordbatch-cha…
Browse files Browse the repository at this point in the history
…nges

Document recordbatch offset changes from #1032
  • Loading branch information
eapache authored Feb 16, 2018
2 parents bfaa60f + 97f65c7 commit 3d1fae1
Showing 1 changed file with 7 additions and 0 deletions.
7 changes: 7 additions & 0 deletions produce_set.go
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,13 @@ func (ps *produceSet) buildRequest() *ProduceRequest {
for topic, partitionSet := range ps.msgs {
for partition, set := range partitionSet {
if req.Version >= 3 {
// If the API version we're hitting is 3 or greater, we need to calculate
// offsets for each record in the batch relative to FirstOffset.
// Additionally, we must set LastOffsetDelta to the value of the last offset
// in the batch. Since the OffsetDelta of the first record is 0, we know that the
// final record of any batch will have an offset of (# of records in batch) - 1.
// (See https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-Messagesets
// under the RecordBatch section for details.)
rb := set.recordsToSend.RecordBatch
if len(rb.Records) > 0 {
rb.LastOffsetDelta = int32(len(rb.Records) - 1)
Expand Down

0 comments on commit 3d1fae1

Please sign in to comment.