@SuppressWarnings("unchecked")
    private void reprocessState(final List<TopicPartition> topicPartitions,
                                final Map<TopicPartition, Long> highWatermarks,
                                final InternalTopologyBuilder.ReprocessFactory reprocessFactory,
                                final String storeName) {
        final Processor source = reprocessFactory.processorSupplier().get();
        source.init(globalProcessorContext);

        for (final TopicPartition topicPartition : topicPartitions) {
            long currentDeadline = NO_DEADLINE;

            globalConsumer.assign(Collections.singletonList(topicPartition));
            long offset;
            final Long checkpoint = checkpointFileCache.get(topicPartition);
            if (checkpoint != null) {
                globalConsumer.seek(topicPartition, checkpoint);
                offset = checkpoint;
            } else {
                globalConsumer.seekToBeginning(Collections.singletonList(topicPartition));
                offset = getGlobalConsumerOffset(topicPartition);
            }
            final Long highWatermark = highWatermarks.get(topicPartition);
            stateRestoreListener.onRestoreStart(topicPartition, storeName, offset, highWatermark);

            long restoreCount = 0L;

            while (offset < highWatermark) {
                // we add `request.timeout.ms` to `poll.ms` because `poll.ms` might be too short
                // to give a fetch request a fair chance to actually complete and we don't want to
                // start `task.timeout.ms` too early
                //
                // TODO with https://issues.apache.org/jira/browse/KAFKA-10315 we can just call
                //      `poll(pollMS)` without adding the request timeout and do a more precise
                //      timeout handling
                final ConsumerRecords<byte[], byte[]> records = globalConsumer.poll(pollMsPlusRequestTimeout);
                if (records.isEmpty()) {
                    currentDeadline = maybeUpdateDeadlineOrThrow(currentDeadline);
                } else {
                    currentDeadline = NO_DEADLINE;
                }

                for (final ConsumerRecord<byte[], byte[]> record : records.records(topicPartition)) {
                    final ProcessorRecordContext recordContext =
                        new ProcessorRecordContext(
                            record.timestamp(),
                            record.offset(),
                            record.partition(),
                            record.topic(),
                            record.headers(),
                            record);
                    globalProcessorContext.setRecordContext(recordContext);

                    try {
                        if (record.key() != null) {
                            source.process(new Record<>(
                                reprocessFactory.keyDeserializer().deserialize(record.topic(), record.key()),
                                reprocessFactory.valueDeserializer().deserialize(record.topic(), record.value()),
                                record.timestamp(),
                                record.headers()));
                            restoreCount++;
                        }
                    } catch (final Exception deserializationException) {
                        handleDeserializationFailure(
                            deserializationExceptionHandler,
                            globalProcessorContext,
                            deserializationException,
                            record,
                            log,
                            droppedRecordsSensor(
                                Thread.currentThread().getName(),
                                globalProcessorContext.taskId().toString(),
                                globalProcessorContext.metrics()
                            )
                        );
                    }
                }

                offset = getGlobalConsumerOffset(topicPartition);

                stateRestoreListener.onBatchRestored(topicPartition, storeName, offset, restoreCount);
            }
            stateRestoreListener.onRestoreEnd(topicPartition, storeName, restoreCount);
            checkpointFileCache.put(topicPartition, offset);

        }
    }
This class is responsible for the initialization, restoration, closing, flushing etc
of Global State Stores. There is only ever 1 instance of this class per Application Instance.
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java