### Question:
ProcessingPartition implements Closeable { public void load(List<ConsumerRecord<K, V>> records) { List<ConsumerRecord<K, V>> recordsToAdd = new ArrayList<>(); LOGGER.debug("Loading records for partition [{}]", topicPartition); if (hasNextRecord()) { LOGGER.debug("Combining existing records with added records for partition [{}]", topicPartition); this.records.forEachRemaining(recordsToAdd::add); } recordsToAdd.addAll(records); this.records = recordsToAdd.iterator(); } ProcessingPartition(TopicPartition topicPartition, ProcessingConfig config, Consumer<K, V> consumer); boolean ack(long offset); boolean fail(long offset); void maybeUnpause(long currentTime); OffsetAndMetadata getCommittableOffset(); long getCommittableOffsetsSize(); void committedOffset(long committedOffset); void load(List<ConsumerRecord<K, V>> records); ConsumerRecord<K, V> nextRecord(); boolean hasNextRecord(); TopicPartition getTopicPartition(); @Override void close();  }### Answer:
@Test public void load() { partition.load(Arrays.asList(record(1L), record(2L), record(3L))); assertThat(partition.records.hasNext(), is(true)); assertRecordsAreEqual(partition.records.next(), record(1L)); assertThat(partition.records.hasNext(), is(true)); assertRecordsAreEqual(partition.records.next(), record(2L)); assertThat(partition.records.hasNext(), is(true)); assertRecordsAreEqual(partition.records.next(), record(3L)); assertThat(partition.records.hasNext(), is(false)); }
@Test public void load_withExistingRecords() { partition.load(Arrays.asList(record(1L))); partition.load(Arrays.asList(record(2L))); assertThat(partition.records.hasNext(), is(true)); assertRecordsAreEqual(partition.records.next(), record(1L)); assertThat(partition.records.hasNext(), is(true)); assertRecordsAreEqual(partition.records.next(), record(2L)); assertThat(partition.records.hasNext(), is(false)); }