Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java (Java)

Code before:

                final ShareFetch<K, V> fetch = pollForFetches(timer);
                if (!fetch.isEmpty()) {
                    currentFetch = fetch;
                    return new ConsumerRecords<>(fetch.records(), Map.of());
                }

                metadata.maybeThrowAnyException();

                // We will wait for retryBackoffMs
            } while (timer.notExpired());

            return ConsumerRecords.empty();
        } catch (ShareFetchException e) {
            currentFetch = (ShareFetch<K, V>) e.shareFetch();
            throw e.cause();
        } finally {
            kafkaShareConsumerMetrics.recordPollEnd(timer.currentTimeMs());
            release();
        }
    }

    private ShareFetch<K, V> pollForFetches(final Timer timer) {
        long pollTimeout = Math.min(applicationEventHandler.maximumTimeToWait(), timer.remainingMs());

        Map<TopicIdPartition, NodeAcknowledgements> acknowledgementsMap = currentFetch.takeAcknowledgedRecords();

        // If data is available already, return it immediately
        final ShareFetch<K, V> fetch = collect(acknowledgementsMap);
        if (!fetch.isEmpty()) {
            return fetch;
        }

        // Wait a bit - this is where we will fetch records
        Timer pollTimer = time.timer(pollTimeout);
        wakeupTrigger.setShareFetchAction(fetchBuffer);

        try {
            fetchBuffer.awaitNotEmpty(pollTimer);
        } catch (InterruptException e) {
            log.trace("Timeout during fetch", e);
            throw e;
        } finally {
            timer.update(pollTimer.currentTimeMs());
            wakeupTrigger.clearTask();
        }

        return collect(Collections.emptyMap());
    }

    private ShareFetch<K, V> collect(Map<TopicIdPartition, NodeAcknowledgements> acknowledgementsMap) {

Review comment (suggestion):

I find this change make `testShareGroupMaxSizeConfigExceeded` become flaky.
The following code will fix this issue but seem to violate the comments.

```
// Handle any acknowledgements which completed while we were waiting, but do not throw
// the exception because the fetched records would then not be returned to the caller
try {
    handleCompletedAcknowledgements(false);
} catch (GroupMaxSizeReachedException e) {
    log.warn("Exception thrown in acknowledgement commit callback", e);
    throw e;
} catch (Throwable t) {
    log.warn("Exception thrown in acknowledgement commit callback", t);
}
```

Code after:

                final ShareFetch<K, V> fetch = pollForFetches(timer);
                if (!fetch.isEmpty()) {
                    currentFetch = fetch;
                    return new ConsumerRecords<>(fetch.records(), Map.of());
                }

                metadata.maybeThrowAnyException();

                // We will wait for retryBackoffMs
            } while (timer.notExpired());

            return ConsumerRecords.empty();
        } catch (ShareFetchException e) {
            currentFetch = (ShareFetch<K, V>) e.shareFetch();
            throw e.cause();
        } finally {
            kafkaShareConsumerMetrics.recordPollEnd(timer.currentTimeMs());

            // Handle any acknowledgements which completed while we were waiting, but do not throw
            // the exception because the fetched records would then not be returned to the caller
            try {
                handleCompletedAcknowledgements(false);
            } catch (Throwable t) {
                log.warn("Exception thrown in acknowledgement commit callback", t);
            }

            release();
        }
    }

    private ShareFetch<K, V> pollForFetches(final Timer timer) {
        long pollTimeout = Math.min(applicationEventHandler.maximumTimeToWait(), timer.remainingMs());

        Map<TopicIdPartition, NodeAcknowledgements> acknowledgementsMap = currentFetch.takeAcknowledgedRecords();

        // If data is available already, return it immediately
        final ShareFetch<K, V> fetch = collect(acknowledgementsMap);
        if (!fetch.isEmpty()) {
            return fetch;
        }

        // Wait a bit - this is where we will fetch records
        Timer pollTimer = time.timer(pollTimeout);
        wakeupTrigger.setShareFetchAction(fetchBuffer);

        try {
            fetchBuffer.awaitNotEmpty(pollTimer);
        } catch (InterruptException e) {
            log.trace("Timeout during fetch", e);
            throw e;