@Override
    public String toString() {
        return "EmbeddedRocksDBStateBackend{"
                + ", localRocksDbDirectories="
                + Arrays.toString(localRocksDbDirectories)
                + ", enableIncrementalCheckpointing="
                + enableIncrementalCheckpointing
                + ", numberOfTransferThreads="
                + numberOfTransferThreads
                + ", writeBatchSize="
                + writeBatchSize
                + '}';
    }
@TestTemplate
    public void testSmallFilesCompaction() throws Exception {
        ValueStateDescriptor<String> kvId = new ValueStateDescriptor<>("id", String.class);
        SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
        CheckpointStreamFactory streamFactory = createStreamFactory();

        final KeyGroupRange range = KeyGroupRange.of(0, 49);
        double expectedNumSstFiles = range.getNumberOfKeyGroups() * .5;
        final CheckpointableKeyedStateBackend<Integer> backend =
                createKeyedBackend(
                        IntSerializer.INSTANCE,
                        range.getEndKeyGroup() - range.getStartKeyGroup() + 1,
                        range,
                        env);
        try {
            ValueState<String> state =
                    backend.getPartitionedState(
                            VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);

            for (int i = range.getStartKeyGroup(); i < range.getEndKeyGroup(); i++) {
                backend.setCurrentKey(i);
                state.update(Integer.toString(i));
                // snapshot to force flushing memtables to disk and create a new SST file
                runSnapshot(
                        backend.snapshot(
                                i, // checkpoint id
                                i, // timestamp
                                streamFactory,
                                CheckpointOptions.forCheckpointWithDefaultLocation()),
                        sharedStateRegistry);
            }

            // expect files under dpPath: job_123_op_456/db/*.sst
            File sstPath = new File(dbPath).listFiles()[0].listFiles()[0];

            int length = sstPath.listFiles((dir, name) -> name.endsWith(".sst")).length;
            assertThat(length)
                    .isLessThanOrEqualTo((int) expectedNumSstFiles)
                    .withFailMessage("actual: " + length + ", expected: " + expectedNumSstFiles);

        } finally {
            IOUtils.closeQuietly(backend);
            backend.dispose();
        }
        // allow some time for the background compaction to fail hard by calling closed db
        Thread.sleep(100);
    }