@Override
    void setPartitionTime(final TopicPartition partition, final long partitionTime) {
        final RecordQueue queue = partitionQueues.get(partition);
        if (queue == null) {
            throw new IllegalStateException("Partition " + partition + " not found.");
        }
        if (streamTime < partitionTime) {
            streamTime = partitionTime;
        }
        queue.setPartitionTime(partitionTime);
    }
@Test
    public void shouldThrowIllegalStateExceptionUponSetPartitionTimestampIfPartitionUnknown() {
        final PartitionGroup group = getBasicGroup();

        final IllegalStateException exception = assertThrows(
            IllegalStateException.class,
            () -> group.setPartitionTime(unknownPartition, 0L));
        assertThat(errMessage, equalTo(exception.getMessage()));
    }