@Override
    public void init(final StateStoreContext context, final StateStore root) {
        final boolean consistencyEnabled = StreamsConfig.InternalConfig.getBoolean(
            context.appConfigs(),
            IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED,
            false
        );
        // register the store
        context.register(
            root,
            (RecordBatchingStateRestoreCallback) records -> {
                restoring = true;
                synchronized (position) {
                    for (final ConsumerRecord<byte[], byte[]> record : records) {
                        put(Bytes.wrap(record.key()), record.value());
                        ChangelogRecordDeserializationHelper.applyChecksAndUpdatePosition(
                            record,
                            consistencyEnabled,
                            position
                        );
                    }
                }
                restoring = false;
            }
        );
        this.context = context;
    }
An in-memory LRU cache store based on HashSet and HashMap.
streams/src/main/java/org/apache/kafka/streams/state/internals/MemoryLRUCache.java