public UUID initializeProcessId() {
        if (!hasPersistentStores) {
            final UUID processId = UUID.randomUUID();
            log.info("Created new process id: {}", processId);
            return processId;
        }

        if (!lockStateDirectory()) {
            log.error("Unable to obtain lock as state directory is already locked by another process");
            throw new StreamsException(String.format("Unable to initialize state, this can happen if multiple instances of " +
                                           "Kafka Streams are running in the same state directory " +
                                           "(current state directory is [%s]", stateDir.getAbsolutePath()));
        }

        final File processFile = new File(stateDir, PROCESS_FILE_NAME);
        final ObjectMapper mapper = new ObjectMapper();

        try {
            if (processFile.exists()) {
                try {
                    final StateDirectoryProcessFile processFileData = mapper.readValue(processFile, StateDirectoryProcessFile.class);
                    log.info("Reading UUID from process file: {}", processFileData.processId);
                    if (processFileData.processId != null) {
                        return processFileData.processId;
                    }
                } catch (final Exception e) {
                    log.warn("Failed to read json process file", e);
                }
            }

            final StateDirectoryProcessFile processFileData = new StateDirectoryProcessFile(UUID.randomUUID());
            log.info("No process id found on disk, got fresh process id {}", processFileData.processId);

            mapper.writeValue(processFile, processFileData);
            return processFileData.processId;
        } catch (final IOException e) {
            log.error("Unable to read/write process file due to unexpected exception", e);
            throw new ProcessorStateException(e);
        }
    }
@Test
    public void shouldReadFutureProcessFileFormat() throws Exception {
        final File processFile = new File(appDir, PROCESS_FILE_NAME);
        final ObjectMapper mapper = new ObjectMapper();
        final UUID processId = UUID.randomUUID();
        mapper.writeValue(processFile, new FutureStateDirectoryProcessFile(processId, "some random junk"));

        assertThat(directory.initializeProcessId(), equalTo(processId));
    }