public synchronized void start() throws IllegalStateException, StreamsException {
        if (setState(State.REBALANCING)) {
            log.debug("Starting Streams client");

            if (globalStreamThread != null) {
                globalStreamThread.start();
            }

            final int numThreads = processStreamThread(StreamThread::start);

            log.info("Started {} stream threads", numThreads);

            final Long cleanupDelay = applicationConfigs.getLong(StreamsConfig.STATE_CLEANUP_DELAY_MS_CONFIG);
            stateDirCleaner.scheduleAtFixedRate(() -> {
                // we do not use lock here since we only read on the value and act on it
                if (state == State.RUNNING) {
                    stateDirectory.cleanRemovedTasks(cleanupDelay);
                }
            }, cleanupDelay, cleanupDelay, TimeUnit.MILLISECONDS);

            final long recordingDelay = 0;
            final long recordingInterval = 1;
            if (rocksDBMetricsRecordingService != null) {
                rocksDBMetricsRecordingService.scheduleAtFixedRate(
                    streamsMetrics.rocksDBMetricsRecordingTrigger(),
                    recordingDelay,
                    recordingInterval,
                    TimeUnit.MINUTES
                );
            }
        } else {
            throw new IllegalStateException("The client is either already started or already stopped, cannot re-start");
        }
    }
@Test
    public void shouldCleanupOldStateDirs() {
        prepareStreams();
        prepareStreamThread(streamThreadOne, 1);
        prepareStreamThread(streamThreadTwo, 2);

        try (final MockedStatic<Executors> executorsMockedStatic = mockStatic(Executors.class)) {
            final ScheduledExecutorService cleanupSchedule = mock(ScheduledExecutorService.class);
            executorsMockedStatic.when(() -> Executors.newSingleThreadScheduledExecutor(
                any(ThreadFactory.class)
            )).thenReturn(cleanupSchedule);

            try (MockedConstruction<StateDirectory> ignored = mockConstruction(StateDirectory.class,
                (mock, context) -> when(mock.initializeProcessId()).thenReturn(UUID.randomUUID()))) {
                props.setProperty(StreamsConfig.STATE_CLEANUP_DELAY_MS_CONFIG, "1");
                final StreamsBuilder builder = new StreamsBuilder();
                builder.table("topic", Materialized.as("store"));

                try (final KafkaStreams streams = new KafkaStreams(builder.build(), props, supplier, time)) {
                    streams.start();
                }
            }

            verify(cleanupSchedule).scheduleAtFixedRate(any(Runnable.class), eq(1L), eq(1L), eq(TimeUnit.MILLISECONDS));
            verify(cleanupSchedule).shutdownNow();
        }
    }