apoorvmittal10 commented on PR #23911:
URL: https://github.com/apache/kafka/pull/23911#issuecomment-6365229941
> Thanks for the changes. However, I think this behaviour is introducing
another case of starvation. The case isn't a very regular way of share fetch
but it can occur and trunk was protected earlier, so seems like a regression.
The case is when a share fetch hands its record budget to partitions in
response order, so the first partition with enough data takes all of it. That
order comes from two stacked mechanisms:
>
> 1. LRU order: the share session moves the partition it just served to the
end of its list.
> 2. Epoch rotation: the broker then rotates the list by `sessionEpoch %
numPartitions`.
>
> The client advances the session epoch on every ShareFetch and every
ShareAcknowledge. In a `poll(); commitSync();` loop the epoch therefore moves
by 2 per fetch, so the rotation never changes. If the epoch is odd, rotation 1
puts the just-served partition first on every fetch with 2 partitions. That
partition takes the whole budget each time, and the other starves for as long
as the first has at least one fetch’s worth of records. With 4 partitions and
odd parity, one partition is never served at all.
>
> A single unpaired request is enough to make the epoch odd:
>
> an empty fetch (for example, before the share partitions finish
initializing, or during lock contention with another member), or an extra
acknowledge, which is how the test triggers it.
>
> **Why trunk doesn’t starve:** the old fetch-offset cache on SharePartition
happened to put the just-served partition last in the response, so the rotation
never mattered. This PR reads every partition up front, so the response follows
the rotated order and that accidental protection is gone.
>
> Here is the test case you can use in `ShareConsumerFairnessTest.java` to
prove it -
>
> ```
> /**
> * In a poll and commit loop every share fetch is followed by one
share acknowledge, and both advance the share
> * session epoch, hence the epoch moves by two between share fetches
and the rotation of the partitions in the
> * share fetch does not change. A single unpaired request, such as a
share fetch which returns no records or, as
> * in this test, the acknowledgements of one poll committed in two
requests, flips the parity of the epoch. The
> * parity before the flip is not known upfront, hence both partitions
must be served fairly in both phases.
> */
> @ClusterTest
> public void
testPartitionsServedFairlyWhenShareSessionEpochParityChanges() throws Exception
{
> String topic = "fairness-epoch-parity";
> String groupId = "fairness-epoch-parity-group";
> createTopic(topic, PARTITIONS, 1);
> produceRecords(topic);
> alterShareAutoOffsetReset(groupId, "earliest");
>
> Map<Integer, Integer> beforeParityChange;
> Map<Integer, Integer> afterParityChange;
> try (ShareConsumer<byte[], byte[]> consumer =
createShareConsumer(groupId, Map.of(
> ConsumerConfig.MAX_POLL_RECORDS_CONFIG,
String.valueOf(MAX_POLL_RECORDS),
> ConsumerConfig.SHARE_ACQUIRE_MODE_CONFIG, "record_limit",
> ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
"explicit"))) {
> consumer.subscribe(Set.of(topic));
> beforeParityChange = pollAndCommit(consumer);
>
> // Commit the acknowledgements of one poll in two requests,
which flips the parity of the epoch.
> List<ConsumerRecord<byte[], byte[]>> records =
pollRecords(consumer);
> int half = records.size() / 2;
> records.subList(0, half).forEach(consumer::acknowledge);
> consumer.commitSync();
> records.subList(half,
records.size()).forEach(consumer::acknowledge);
> consumer.commitSync();
>
> afterParityChange = pollAndCommit(consumer);
> }
>
> // Each phase consumes far fewer records than the backlog of
either partition, hence both partitions have
> // records throughout and each should get a meaningful share of
the records of every phase.
> String recordsByPhase = "records by partition before the epoch
parity changed: " + beforeParityChange
> + ", after: " + afterParityChange;
> for (Map<Integer, Integer> phase : List.of(beforeParityChange,
afterParityChange)) {
> int phaseRecords =
phase.values().stream().mapToInt(Integer::intValue).sum();
> for (int p = 0; p < PARTITIONS; p++) {
> assertTrue(phase.getOrDefault(p, 0) * 5 >= phaseRecords,
> "Partition " + p + " was starved, " + recordsByPhase);
> }
> }
> }
>
> private void produceRecords(String topic) {
> try (Producer<byte[], byte[]> producer = createProducer()) {
> for (int i = 0; i < RECORDS_PER_PARTITION; i++) {
> for (int p = 0; p < PARTITIONS; p++) {
> producer.send(new ProducerRecord<>(topic, p, null,
null, "value".getBytes()));
> }
> }
> producer.flush();
> }
> }
>
> /**
> * Polls until {@link #POLLS_PER_PHASE} polls have returned records,
and commits the acknowledgements of each poll
> * in one request.
> */
> private Map<Integer, Integer> pollAndCommit(ShareConsumer<byte[],
byte[]> consumer) {
> Map<Integer, Integer> recordsByPartition = new TreeMap<>();
> for (int poll = 0; poll < POLLS_PER_PHASE; poll++) {
> for (ConsumerRecord<byte[], byte[]> record :
pollRecords(consumer)) {
> recordsByPartition.merge(record.partition(), 1,
Integer::sum);
> consumer.acknowledge(record);
> }
> consumer.commitSync();
> }
> return recordsByPartition;
> }
>
> /**
> * Polls until records are returned, the test timeout fails the test
if no records are delivered.
> */
> private List<ConsumerRecord<byte[], byte[]>>
pollRecords(ShareConsumer<byte[], byte[]> consumer) {
> List<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
> while (records.isEmpty()) {
> consumer.poll(Duration.ofMillis(500)).forEach(records::add);
> }
> return records;
> }
> ```
This is a valid case and thanks for highlighting. Seems a bit niche and it
requires to change session handling and partition rotate strategy. I am
thinking to have it separately so other PR can be unlocked, I have created jira
for same: https://issues.apache.org/jira/browse/KAFKA-21344
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
