Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java (Java)

Code before:
                } else {
                    log.debug("[GroupId {}] Cannot remove the process ID {} from {} because it does not have any processId",
                            groupId, expectedProcessId, subtopologyId);
                    return partitionsOrNull;
                }
            });
        });
    }

    /**
     * Removes the task process IDs based on the provided assignment.
     *
     * @param assignment    The assignment.
     * @param processIdToRemove The expected process ID.
     * package-private for testing.
     */
    private void removeTaskProcessIdsFromSet(
        Map<String, Set<Integer>> assignment,
        TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentTasksProcessId,
        String processIdToRemove
    ) {
        assignment.forEach((subtopologyId, assignedPartitions) -> {
            currentTasksProcessId.compute(subtopologyId, (__, partitionsOrNull) -> {
                if (partitionsOrNull != null) {
                    assignedPartitions.forEach(partitionId -> {
                        if (!partitionsOrNull.get(partitionId).remove(processIdToRemove)) {
                            log.debug("[GroupId {}] Cannot remove the process ID {} from task {}_{} because the task is " +
                                    "not owned by this process ID", groupId, processIdToRemove, subtopologyId, partitionId);
                        }
                    });
                    if (partitionsOrNull.isEmpty()) {
                        return null;
                    } else {
                        return partitionsOrNull;
                    }
                } else {
                    log.debug("[GroupId {}] Cannot remove the process ID {} from {} because it does not have any process ID",
                            groupId, processIdToRemove, subtopologyId);
                    return partitionsOrNull;
                }
            });
        });
    }

    /**
     * Adds the partition epoch based on the provided assignment.
     *
     * @param tasks     The assigned tasks.
     * @param processId The process ID.
     * package-private for testing.
     */

Review comment (suggestion):
```suggestion
                        if (!partitionsOrNull.containsKey(parititionId) || !partitionsOrNull.get(partitionId).remove(processIdToRemove)) {
```

Code after:
                } else {
                    log.debug("[GroupId {}] Cannot remove the process ID {} from {} because it does not have any processId",
                            groupId, expectedProcessId, subtopologyId);
                    return partitionsOrNull;
                }
            });
        });
    }

    /**
     * Removes the task process IDs based on the provided assignment.
     *
     * @param assignment    The assignment.
     * @param processIdToRemove The expected process ID.
     * package-private for testing.
     */
    private void removeTaskProcessIdsFromSet(
        Map<String, Set<Integer>> assignment,
        TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentTasksProcessId,
        String processIdToRemove
    ) {
        assignment.forEach((subtopologyId, assignedPartitions) -> {
            currentTasksProcessId.compute(subtopologyId, (__, partitionsOrNull) -> {
                if (partitionsOrNull != null) {
                    assignedPartitions.forEach(partitionId -> {
                        if (!partitionsOrNull.containsKey(partitionId) || !partitionsOrNull.get(partitionId).remove(processIdToRemove)) {
                            log.debug("[GroupId {}] Cannot remove the process ID {} from task {}_{} because the task is " +
                                    "not owned by this process ID", groupId, processIdToRemove, subtopologyId, partitionId);
                        }
                    });
                    if (partitionsOrNull.isEmpty()) {
                        return null;
                    } else {
                        return partitionsOrNull;
                    }
                } else {
                    log.debug("[GroupId {}] Cannot remove the process ID {} from {} because it does not have any process ID",
                            groupId, processIdToRemove, subtopologyId);
                    return partitionsOrNull;
                }
            });
        });
    }

    /**
     * Adds the partition epoch based on the provided assignment.
     *
     * @param tasks     The assigned tasks.
     * @param processId The process ID.
     * package-private for testing.
     */