| 631 | } |
| 632 | |
| 633 | func (r *FetchResponse) AddMessageWithTimestamp(topic string, partition int32, key, value Encoder, offset int64, timestamp time.Time, version int8) { |
| 634 | frb := r.getOrCreateBlock(topic, partition) |
| 635 | kb, vb := encodeKV(key, value) |
| 636 | if r.LogAppendTime { |
| 637 | timestamp = r.Timestamp |
| 638 | } |
| 639 | msg := &Message{Key: kb, Value: vb, LogAppendTime: r.LogAppendTime, Timestamp: timestamp, Version: version} |
| 640 | msgBlock := &MessageBlock{Msg: msg, Offset: offset} |
| 641 | if len(frb.RecordsSet) == 0 { |
| 642 | records := newLegacyRecords(&MessageSet{}) |
| 643 | frb.RecordsSet = []*Records{&records} |
| 644 | } |
| 645 | set := frb.RecordsSet[0].MsgSet |
| 646 | set.Messages = append(set.Messages, msgBlock) |
| 647 | } |
| 648 | |
| 649 | func (r *FetchResponse) AddRecordWithTimestamp(topic string, partition int32, key, value Encoder, offset int64, timestamp time.Time) { |
| 650 | frb := r.getOrCreateBlock(topic, partition) |