@Override
    public void completeRestoration(final java.util.function.Consumer<Set<TopicPartition>> offsetResetter) {
        switch (state()) {
            case RUNNING:
                return;

            case RESTORING:
                resetOffsetsIfNeededAndInitializeMetadata(offsetResetter);
                initializeTopology();
                processorContext.initialize();
                if (!eosEnabled) {
                    maybeCheckpoint(true);
                }

                transitionTo(State.RUNNING);

                log.info("Restored and ready to run");

                break;

            case CREATED:
            case SUSPENDED:
            case CLOSED:
                throw new IllegalStateException("Illegal state " + state() + " while completing restoration for active task " + id);

            default:
                throw new IllegalStateException("Unknown state " + state() + " while completing restoration for active task " + id);
        }
    }
@Test
    public void shouldThrowStreamsExceptionWhenFetchCommittedFailed() {
        when(stateManager.taskId()).thenReturn(taskId);
        when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
        final Consumer<byte[], byte[]> consumer = new MockConsumer<byte[], byte[]>(OffsetResetStrategy.EARLIEST) {
            @Override
            public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions) {
                throw new KafkaException("KABOOM!");
            }
        };

        task = createOptimizedStatefulTask(createConfig("100"), consumer);

        task.transitionTo(RESTORING);

        assertThrows(StreamsException.class, () -> task.completeRestoration(noOpResetter -> { }));
    }