Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java (Java)

Code before:
                    return Collections.emptyMap();
                }

            case CLOSED:
                throw new IllegalStateException("Illegal state " + state() + " while preparing active task " + id + " for committing");

            default:
                throw new IllegalStateException("Unknown state " + state() + " while preparing active task " + id + " for committing");
        }
    }

    private Optional<OffsetAndMetadata> findOffsetAndMetadata(final TopicPartition partition) {
        Long offset = partitionGroup.headRecordOffset(partition);
        Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
        final long partitionTime = partitionGroup.partitionTimestamp(partition);
        if (offset == null) {
            if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
                final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
                offset = offsetAndMetadata.offset();
                leaderEpoch = offsetAndMetadata.leaderEpoch();
            } else {
                // it may be that we have not yet consumed any record from this partition, hence nothing to commit
                return Optional.empty();
            }
        }
        return Optional.of(new OffsetAndMetadata(offset,
                leaderEpoch,
                new TopicPartitionMetadata(partitionTime, processorContext.processorMetadata()).encode()));
    }

    private Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
        switch (state()) {
            case CREATED:
            case RESTORING:
                return Collections.emptyMap();

            case RUNNING:
            case SUSPENDED:

                // If there's processor metadata to be committed. We need to commit them to all
                // input partitions
                final Set<TopicPartition> partitionsNeedCommit = processorContext.processorMetadata().needsCommit() ?
                    inputPartitions() : consumedOffsets.keySet();

                return partitionsNeedCommit.stream()
                        .flatMap(partition -> findOffsetAndMetadata(partition)
                                .map(offsetAndMetadata -> Map.entry(partition, offsetAndMetadata))
                                .stream())
                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

            case CLOSED:

Review comment (refactor):
nit: should we move this into the ` if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {` block?

Code after:
                    return Collections.emptyMap();
                }

            case CLOSED:
                throw new IllegalStateException("Illegal state " + state() + " while preparing active task " + id + " for committing");

            default:
                throw new IllegalStateException("Unknown state " + state() + " while preparing active task " + id + " for committing");
        }
    }

    private Optional<OffsetAndMetadata> findOffsetAndMetadata(final TopicPartition partition) {
        Long offset = partitionGroup.headRecordOffset(partition);
        Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
        final long partitionTime = partitionGroup.partitionTimestamp(partition);
        if (offset == null) {
            final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
            if (offsetAndMetadata == null) {
                // it may be that we have not yet consumed any record from this partition, hence nothing to commit
                return Optional.empty();
            }
            offset = offsetAndMetadata.offset();
            leaderEpoch = offsetAndMetadata.leaderEpoch();
        }
        return Optional.of(new OffsetAndMetadata(offset,
                leaderEpoch,
                new TopicPartitionMetadata(partitionTime, processorContext.processorMetadata()).encode()));
    }

    private Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
        switch (state()) {
            case CREATED:
            case RESTORING:
                return Collections.emptyMap();

            case RUNNING:
            case SUSPENDED:

                // If there's processor metadata to be committed. We need to commit them to all
                // input partitions
                final Set<TopicPartition> partitionsNeedCommit = processorContext.processorMetadata().needsCommit() ?
                    inputPartitions() : consumedOffsets.keySet();

                return partitionsNeedCommit.stream()
                        .map(partition -> findOffsetAndMetadata(partition)
                                .map(offsetAndMetadata -> Map.entry(partition, offsetAndMetadata)))
                        .filter(Optional::isPresent)
                        .map(Optional::get)
                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

            case CLOSED: