class AssignedStreamsTasks extends AssignedTasks<StreamTask> implements RestoringTasks {

    boolean maybeResumeSuspendedTask(final TaskId taskId,
                                     final Set<TopicPartition> partitions) {
        if (suspended.containsKey(taskId)) {
            final StreamTask task = suspended.get(taskId);
            log.trace("Found suspended stream task {}", taskId);
            removeTaskFromAllStateMaps(task, Collections.emptyMap());

            if (task.partitions().equals(partitions)) {
                task.resume();
                try {
                    transitionToRunning(task);
                } catch (final TaskMigratedException e) {
                    // we need to catch migration exception internally since this function
                    // is triggered in the rebalance callback
                    log.info("Failed to resume stream task {} since it got migrated to another thread already. " +
                             "Will trigger a new rebalance and close all tasks as zombies together.", task.id());
                    throw e;
                }
                log.trace("Resuming the suspended stream task {}", task.id());
                return true;
            } else {
                log.warn("Couldn't resume stream task {} assigned partitions {}, task partitions {}", taskId, partitions, task.partitions());
                task.closeSuspended(true, null);
            }
        }
        return false;
    }

      AssignedStreamsTasks(final LogContext logContext);

}