AddRecordBatchWithTimestamp is similar to AddRecordWithTimestamp But instead of appending 1 record to a batch, it append a new batch containing 1 record to the fetchResponse Since transaction are handled on batch level (the whole batch is either committed or aborted), use this to test transactions
(topic string, partition int32, key, value Encoder, offset int64, producerID int64, isTransactional bool, timestamp time.Time)
| 662 | // But instead of appending 1 record to a batch, it append a new batch containing 1 record to the fetchResponse |
| 663 | // Since transaction are handled on batch level (the whole batch is either committed or aborted), use this to test transactions |
| 664 | func (r *FetchResponse) AddRecordBatchWithTimestamp(topic string, partition int32, key, value Encoder, offset int64, producerID int64, isTransactional bool, timestamp time.Time) { |
| 665 | frb := r.getOrCreateBlock(topic, partition) |
| 666 | kb, vb := encodeKV(key, value) |
| 667 | |
| 668 | records := newDefaultRecords(&RecordBatch{Version: 2, LogAppendTime: r.LogAppendTime, FirstTimestamp: timestamp, MaxTimestamp: r.Timestamp}) |
| 669 | batch := &RecordBatch{ |
| 670 | Version: 2, |
| 671 | LogAppendTime: r.LogAppendTime, |
| 672 | FirstTimestamp: timestamp, |
| 673 | MaxTimestamp: r.Timestamp, |
| 674 | FirstOffset: offset, |
| 675 | LastOffsetDelta: 0, |
| 676 | ProducerID: producerID, |
| 677 | IsTransactional: isTransactional, |
| 678 | } |
| 679 | rec := &Record{Key: kb, Value: vb, OffsetDelta: 0, TimestampDelta: timestamp.Sub(batch.FirstTimestamp)} |
| 680 | batch.addRecord(rec) |
| 681 | records.RecordBatch = batch |
| 682 | |
| 683 | frb.RecordsSet = append(frb.RecordsSet, &records) |
| 684 | } |
| 685 | |
| 686 | func (r *FetchResponse) AddControlRecordWithTimestamp(topic string, partition int32, offset int64, producerID int64, recordType ControlRecordType, timestamp time.Time) { |
| 687 | frb := r.getOrCreateBlock(topic, partition) |
no test coverage detected