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);
        }
    }
@return true if the state directory was successfully locked
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java