KAFKA-9686 MockConsumer#endOffsets should be idempotent

[chia7712]

    private Long getEndOffset(List<Long> offsets) {
        if (offsets == null || offsets.isEmpty()) {
            return null;
        }
        return offsets.size() > 1 ? offsets.remove(0) : offsets.get(0);
    }

The above code has two issues.

1. It does not return the latest offset since the latest offset is at the end of offsets
2. It removes the element from offsets so MockConsumer#endOffsets gets non-idempotent

https://issues.apache.org/jira/browse/KAFKA-9686

The following flaky is fixed by this PR

1. KafkaBasedLogTest.testSendAndReadToEnd

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[chia7712]

retest this please

[hachikuji]

The code seems intentionally written to store a sequence of end offsets so that each call advances the end offset. If we do not need that capability, then perhaps we can change the type of endOffsets and simplify the usage.

cc @bbejeck Looks like you added this logic. Do you know if we rely on it?

[kkonstantine]

I agree with @hachikuji 's observation. This test broke when #8220 retired the innerUpdateEndOffsets and replaced the two public methods addEndOffsets and updateEndOffsets that were both using innerUpdateEndOffsets with a single updateEndOffsets. However the version that was kept was actually what the unused addEndOffsets was doing (appending to the end offsets) and not what updateEndOffsets described as an overwrite of the value in endOffsets.

Thus, probably the issue should be fixed in updateEndOffsets to keep meaning overwrite. Unless we find a usage for addEndOffsets, in which case we might need some more adjustments too. Of course, if we keep only updateEndOffsets that overwrites the value of the map, we might as well simplify getEndOffset. I haven't tracked when and why addEndOffsets was stopped being used.

[chia7712]

@hachikuji @kkonstantine thanks for reviews.

Will address your comment :)

[kkonstantine]

Added also a comment inline, after tracking that this test broke with a consolidation in: #8220

[kkonstantine]

I agree with @hachikuji 's observation. This test broke when #8220 retired the innerUpdateEndOffsets and replaced the two public methods addEndOffsets and updateEndOffsets that were both using innerUpdateEndOffsets with a single updateEndOffsets. However the version that was kept was actually what the unused addEndOffsets was doing (appending to the end offsets) and not what updateEndOffsets described as an overwrite of the value in endOffsets.

Thus, probably the issue should be fixed in updateEndOffsets to keep meaning overwrite. Unless we find a usage for addEndOffsets, in which case we might need some more adjustments too. Of course, if we keep only updateEndOffsets that overwrites the value of the map, we might as well simplify getEndOffset. I haven't tracked when and why addEndOffsets was stopped being used.

[kkonstantine]

This LGTM now.
Thanks for addressing this @chia7712 !

I'm merging to trunk