streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
private void tryToLockAllNonEmptyTaskDirectories() {
        for (final File dir : stateDirectory.listNonEmptyTaskDirectories()) {
            try {
                final TaskId id = TaskId.parse(dir.getName());
                try {
                    if (stateDirectory.lock(id)) {
                        lockedTaskDirectories.add(id);
                        if (!tasks.containsKey(id)) {
                            log.debug("Temporarily locked unassigned task {} for the upcoming rebalance", id);
                        }
                    }
                } catch (final IOException e) {
                    // if for any reason we can't lock this task dir, just move on
                    log.warn(String.format("Exception caught while attempting to lock task %s:", id), e);
                }
            } catch (final TaskIdFormatException e) {
                // ignore any unknown files that sit in the same directory
            }
        }
    }
/**
     * Makes a weak attempt to lock all non-empty task directories in the state dir. We are responsible for computing and
     * reporting the offset sum for any unassigned tasks we obtain the lock for in the upcoming rebalance. Tasks
     * that we locked but didn't own will be released at the end of the rebalance (unless of course we were
     * assigned the task as a result of the rebalance). This method should be idempotent.
     */
Makes a weak attempt to lock all non-empty task directories in the state dir. We are responsible for computing and
reporting the offset sum for any unassigned tasks we obtain the lock for in the upcoming rebalance. Tasks
that we locked but didn't own will be released at the end of the rebalance (unless of course we were
assigned the task as a result of the rebalance). This method should be idempotent.
Makes a weak attempt to lock all non - empty task directories in the state dir . We are responsible for computing and reporting the offset sum for any unassigned tasks we obtain the lock for in the upcoming rebalance . Tasks that we locked but didn ' t own will be released at the end of the rebalance ( unless of course we were assigned the task as a result of the rebalance ) . This method should be idempotent .
private void tryToLockAllNonEmptyTaskDirectories() {
        for (final File dir : stateDirectory.listNonEmptyTaskDirectories()) {
            try {
                final TaskId id = TaskId.parse(dir.getName());
                try {
                    if (stateDirectory.lock(id)) {
                        lockedTaskDirectories.add(id);
                        if (!tasks.containsKey(id)) {
                            log.debug("Temporarily locked unassigned task {} for the upcoming rebalance", id);
                        }
                    }
                } catch (final IOException e) {
                    log.warn(String.format("Exception caught while attempting to lock task %s:", id), e);
                }
            } catch (final TaskIdFormatException e) {
            }
        }
    }
private void tryToLockAllNonEmptyTaskDirectories ( ) { for ( final File dir : stateDirectory . listNonEmptyTaskDirectories ( ) ) { try { final TaskId id = TaskId . parse ( dir . getName ( ) ) ; try { if ( stateDirectory . lock ( id ) ) { lockedTaskDirectories . add ( id ) ; if ( ! tasks . containsKey ( id ) ) { log . debug ( "Temporarily locked unassigned task {} for the upcoming rebalance" , id ) ; } } } catch ( final IOException e ) { log . warn ( String . format ( "Exception caught while attempting to lock task %s:" , id ) , e ) ; } } catch ( final TaskIdFormatException e ) { } } }
Makes a weak attempt to lock all non-empty task directories in the state dir.
Makes a weak attempt to lock all non - empty task directories in the state dir .
// if for any reason we can't lock this task dir, just move on // ignore any unknown files that sit in the same directory