int addRawRecords(final Iterable<ConsumerRecord<byte[], byte[]>> rawRecords) {
        for (final ConsumerRecord<byte[], byte[]> rawRecord : rawRecords) {
            fifoQueue.addLast(rawRecord);
        }

        updateHead();

        return size();
    }
@Test
    public void shouldThrowOnNegativeTimestamp() {
        final List<ConsumerRecord<byte[], byte[]>> records = Collections.singletonList(
            new ConsumerRecord<>("topic", 1, 1, -1L, TimestampType.CREATE_TIME, 0, 0, recordKey, recordValue,
                new RecordHeaders(), Optional.empty()));

        final RecordQueue queue = new RecordQueue(
            new TopicPartition("topic", 1),
            mockSourceNodeWithMetrics,
            new FailOnInvalidTimestamp(),
            new LogAndContinueExceptionHandler(),
            new InternalMockProcessorContext(),
            new LogContext());

        final StreamsException exception = assertThrows(
            StreamsException.class,
            () -> queue.addRawRecords(records)
        );
        assertThat(exception.getMessage(), equalTo("Input record ConsumerRecord(topic = topic, partition = 1, " +
            "leaderEpoch = null, offset = 1, CreateTime = -1, deliveryCount = null, serialized key size = 0, " +
            "serialized value size = 0, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1, value = 10) " +
            "has invalid (negative) timestamp. Possibly because a pre-0.10 producer client was used to write this record " +
            "to Kafka without embedding a timestamp, or because the input topic was created before upgrading the Kafka " +
            "cluster to 0.10+. Use a different TimestampExtractor to process this data."));
    }