| 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) |
| 651 | kb, vb := encodeKV(key, value) |
| 652 | if len(frb.RecordsSet) == 0 { |
| 653 | records := newDefaultRecords(&RecordBatch{Version: 2, LogAppendTime: r.LogAppendTime, FirstTimestamp: timestamp, MaxTimestamp: r.Timestamp}) |
| 654 | frb.RecordsSet = []*Records{&records} |
| 655 | } |
| 656 | batch := frb.RecordsSet[0].RecordBatch |
| 657 | rec := &Record{Key: kb, Value: vb, OffsetDelta: offset, TimestampDelta: timestamp.Sub(batch.FirstTimestamp)} |
| 658 | batch.addRecord(rec) |
| 659 | } |
| 660 | |
| 661 | // AddRecordBatchWithTimestamp is similar to AddRecordWithTimestamp |
| 662 | // But instead of appending 1 record to a batch, it append a new batch containing 1 record to the fetchResponse |