static TaskId parseTaskDirectoryName(final String taskIdStr, final String namedTopology) {
        final int index = taskIdStr.indexOf('_');
        if (index <= 0 || index + 1 >= taskIdStr.length()) {
            throw new TaskIdFormatException(taskIdStr);
        }

        try {
            final int topicGroupId = Integer.parseInt(taskIdStr.substring(0, index));
            final int partition = Integer.parseInt(taskIdStr.substring(index + 1));

            return new TaskId(topicGroupId, partition, namedTopology);
        } catch (final Exception e) {
            throw new TaskIdFormatException(taskIdStr, e);
        }
    }
Parse the task directory name (of the form topicGroupId_partition) and construct the TaskId with the
 optional namedTopology (may be null)

 @throws TaskIdFormatException if the taskIdStr is not a valid {@link TaskId}
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java