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

Code before:
                    hasPendingTxCommit = eosEnabled;

                    log.debug("Prepared {} task for committing", state());
                    return committableOffsetsAndMetadata();
                } else {
                    log.debug("Skipped preparing {} task for commit since there is nothing to commit", state());
                    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 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) {
                try {
                    offset = mainConsumer.position(partition);
                    leaderEpoch = Optional.empty();
                } catch (final TimeoutException error) {
                    // the `consumer.position()` call should never block, because we know that we did process data
                    // for the requested partition and thus the consumer should have a valid local position
                    // that it can return immediately

                    // hence, a `TimeoutException` indicates a bug and thus we rethrow it as fatal `IllegalStateException`
                    throw new IllegalStateException(error);
                } catch (final KafkaException fatal) {
                    throw new StreamsException(fatal);
                }
            } else {
                offset = offsetAndMetadata.offset();
                leaderEpoch = offsetAndMetadata.leaderEpoch();
            }
        }
        return new OffsetAndMetadata(offset,
                leaderEpoch,
                new TopicPartitionMetadata(partitionTime, processorContext.processorMetadata()).encode());
    }

    private Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
        switch (state()) {
            case CREATED:
            case RESTORING:

Review comment (suggestion):
my understanding so far:
we have some metadata that's one per task. but we have multiple input partitions. so instead of picking one, we just add it everywhere. and during restore we just read them all in whatever order and expect them to be more or less the same.
so if we are adding a new partition on the fly, we want that partition to also have that metadata. if we follow that logic.
so if we follow that logic, we need to commit even for that empty partition. 

so I added a fallback to the previous logic when we ask the consumer about the offset





Code after:
                    hasPendingTxCommit = eosEnabled;

                    log.debug("Prepared {} task for committing", state());
                    return committableOffsetsAndMetadata();
                } else {
                    log.debug("Skipped preparing {} task for commit since there is nothing to commit", state());
                    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)