def METHOD_NAME(self, my_task):
    # If the threshold was already reached, there is nothing else to do.
    if my_task._has_state(TaskState.COMPLETED):
        return False
    if my_task._has_state(TaskState.READY):
        return True
    # Retrieve a list of all activated tasks from the associated
    # task that did the conditional parallel split.
    split_task = my_task._find_ancestor_from_name(self.split_task)
    if split_task is None:
        raise WorkflowException(f'Join with %s, which was not reached {self.split_task}', task_spec=self)
    tasks = split_task.task_spec._get_activated_threads(split_task)
    # The default threshold is the number of threads that were started.
    threshold = valueof(my_task, self.threshold)
    if threshold is None:
        threshold = len(tasks)
    # Look up which tasks have already completed.
    waiting_tasks = []
    completed = 0
    for task in tasks:
        # Refresh path prediction.
        task.task_spec._predict(task)
        if self._branch_is_complete(task):
            completed += 1
        else:
            waiting_tasks.append(task)
    # If the threshold was reached, get ready to fire.
    if completed >= threshold:
        # If this is a cancelling join, cancel all incoming branches,
        # except for the one that just completed.
        if self.cancel_remaining:
            for task in waiting_tasks:
                task.cancel()
        return True
    # We do NOT set the task state to COMPLETED, because in
    # case all other incoming tasks get cancelled (or never reach
    # the ThreadMerge for other reasons, such as reaching a stub branch),
    # we need to revisit it.
    return False