public FileRecords.TimestampAndOffset searchForTimestamp(long targetTimestamp, long startingOffset) {
        for (RecordBatch batch : batchesFrom(startingOffset)) {
            if (batch.maxTimestamp() >= targetTimestamp) {
                // We found a message
                for (Record record : batch) {
                    long timestamp = record.timestamp();
                    if (timestamp >= targetTimestamp && record.offset() >= startingOffset)
                        return new FileRecords.TimestampAndOffset(timestamp, record.offset(),
                                maybeLeaderEpoch(batch.partitionLeaderEpoch()));
                }
            }
        }
        return null;
    }
Append records to segment.
Note that lastOffset is the expected value of nextOffset after append. lastOffset = (the real last offset of the
records) + 1

@param records    records to append
@param lastOffset expected next offset after append
@return the size of the appended records
@throws IOException
core/src/main/scala/kafka/log/streamaspect/ElasticLogFileRecords.java