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

        updateHead();

        return size();
    }
@Test
    public void shouldThrowStreamsExceptionWhenKeyDeserializationFails() {
        final byte[] key = Serdes.Long().serializer().serialize("foo", 1L);
        final List<ConsumerRecord<byte[], byte[]>> records = Collections.singletonList(
            new ConsumerRecord<>("topic", 1, 1, 0L, TimestampType.CREATE_TIME, 0, 0, key, recordValue,
                new RecordHeaders(), Optional.empty()));

        final StreamsException exception = assertThrows(
            StreamsException.class,
            () -> queue.addRawRecords(records)
        );
        assertThat(exception.getCause(), instanceOf(SerializationException.class));
    }