Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java (Java)

Code before:
    }

    RocksDBStore(final String name,
                 final String parentDir,
                 final RocksDBMetricsRecorder metricsRecorder) {
        this(name, parentDir, metricsRecorder, true);
    }

    RocksDBStore(final String name,
                 final String parentDir,
                 final RocksDBMetricsRecorder metricsRecorder,
                 final boolean autoManagedIterators) {
        this.name = name;
        this.parentDir = parentDir;
        this.metricsRecorder = metricsRecorder;
        this.autoManagedIterators = autoManagedIterators;
    }

    @Override
    public void init(final StateStoreContext stateStoreContext,
                     final StateStore root) {
        initialized.set(true);
        // open the DB dir
        metricsRecorder.init(metricsImpl(stateStoreContext), stateStoreContext.taskId());
        if (!open) {
            preInit(stateStoreContext);
        }

        addValueProvidersToMetricsRecorder();

        final File positionCheckpointFile = new File(stateStoreContext.stateDir(), name() + ".position");
        this.positionCheckpoint = new OffsetCheckpoint(positionCheckpointFile);
        this.position = StoreQueryUtils.readPositionFromCheckpoint(positionCheckpoint);

        // value getter should always read directly from rocksDB
        // since it is only for values that are already flushed
        this.context = stateStoreContext;
        stateStoreContext.register(
            root,
            (RecordBatchingStateRestoreCallback) this::restoreBatch,
            () -> StoreQueryUtils.checkpointPosition(positionCheckpoint, position)
        );
        consistencyEnabled = StreamsConfig.InternalConfig.getBoolean(
            stateStoreContext.appConfigs(),
            IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED,
            false);
    }

    @Override
    public void preInit(final StateStoreContext stateStoreContext) {
        openDB(stateStoreContext.appConfigs(), stateStoreContext.stateDir());

Review comment (security):
maybe it's not a valid concern, but it looks like this interface would be hard to explain and thus use correctly: there is a two stage init process. sometimes we call preInit(phase one) explicitly, but there are a lot of cases when the author of the store needs to call it themselves in the init(phase two).

as we rely on stateStore.isOpen be true, maybe we should communicate that and follow the same path every time we want to initialize a store?
i.e.
* phase one(preInit/open)
* check if opened
* phase two(init)

it will slightly complicate the implementation on our side, but will make it significantly easier to follow for users

Code after:

    RocksDBStore(final String name,
                 final String parentDir,
                 final RocksDBMetricsRecorder metricsRecorder) {
        this(name, parentDir, metricsRecorder, true);
    }

    RocksDBStore(final String name,
                 final String parentDir,
                 final RocksDBMetricsRecorder metricsRecorder,
                 final boolean autoManagedIterators) {
        this.name = name;
        this.parentDir = parentDir;
        this.metricsRecorder = metricsRecorder;
        this.autoManagedIterators = autoManagedIterators;
    }

    @Override
    public void init(final StateStoreContext stateStoreContext,
                     final StateStore root) {
        // open the DB dir
        metricsRecorder.init(metricsImpl(stateStoreContext), stateStoreContext.taskId());
        openDB(stateStoreContext.appConfigs(), stateStoreContext.stateDir());

        final File positionCheckpointFile = new File(stateStoreContext.stateDir(), name() + ".position");
        this.positionCheckpoint = new OffsetCheckpoint(positionCheckpointFile);
        this.position = StoreQueryUtils.readPositionFromCheckpoint(positionCheckpoint);

        // value getter should always read directly from rocksDB
        // since it is only for values that are already flushed
        this.context = stateStoreContext;
        stateStoreContext.register(
            root,
            (RecordBatchingStateRestoreCallback) this::restoreBatch,
            () -> StoreQueryUtils.checkpointPosition(positionCheckpoint, position)
        );
        consistencyEnabled = StreamsConfig.InternalConfig.getBoolean(
            stateStoreContext.appConfigs(),
            IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED,
            false);
    }

    @SuppressWarnings("unchecked")
    void openDB(final Map<String, Object> configs, final File stateDir) {
        // initialize the default rocksdb options

        final DBOptions dbOptions = new DBOptions();
        final ColumnFamilyOptions columnFamilyOptions = new ColumnFamilyOptions();
        userSpecifiedOptions = new RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter(dbOptions, columnFamilyOptions);

        final BlockBasedTableConfigWithAccessibleCache tableConfig = new BlockBasedTableConfigWithAccessibleCache();