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

Code before:
    private static ShareAcknowledgementMode initializeAcknowledgementMode(ConsumerConfig config, Logger log) {
        String s = config.getString(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG);
        return ShareAcknowledgementMode.fromString(s);
    }

    private void processBackgroundEvents(final Predicate<Exception> ignoreErrorEventException) {
        try {
            processBackgroundEvents();
        } catch (Exception e) {
            if (!ignoreErrorEventException.test(e))
                throw e;
        }
    }

    /**
     * Process the events—if any—that were produced by the {@link ConsumerNetworkThread network thread}.
     * It is possible that {@link ErrorEvent an error}
     * could occur when processing the events. In such cases, the processor will take a reference to the first
     * error, continue to process the remaining events, and then throw the first error that occurred.
     */
    private boolean processBackgroundEvents() {
        AtomicReference<KafkaException> firstError = new AtomicReference<>();

        LinkedList<BackgroundEvent> events = new LinkedList<>();
        backgroundEventQueue.drainTo(events);

        for (BackgroundEvent event : events) {
            try {
                if (event instanceof CompletableEvent)
                    backgroundEventReaper.add((CompletableEvent<?>) event);

                backgroundEventProcessor.process(event);
            } catch (Throwable t) {
                KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);

                if (!firstError.compareAndSet(null, e))
                    log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
            }
        }

        backgroundEventReaper.reap(time.milliseconds());

        if (firstError.get() != null)
            throw firstError.get();

        return !events.isEmpty();
    }

    /**
     * This method can be used by cases where the caller has an event that needs to both block for completion but
     * also process background events. For some events, in order to fully process the associated logic, the

Review comment (suggestion):
`BackgroundEventHandler#drainEvents` is not used, so the metrics `background-event-queue-size` is never reset to zero. One option could be to call `asyncConsumerMetrics.recordBackgroundEventQueueSize(0)` manually. Alternatively, `ShareConsumerImpl` could use `BackgroundEventHandler` instead of `backgroundEventQueue`, similar to `AsyncKafkaConsumer`

Code after:
    private static ShareAcknowledgementMode initializeAcknowledgementMode(ConsumerConfig config, Logger log) {
        String s = config.getString(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG);
        return ShareAcknowledgementMode.fromString(s);
    }

    private void processBackgroundEvents(final Predicate<Exception> ignoreErrorEventException) {
        try {
            processBackgroundEvents();
        } catch (Exception e) {
            if (!ignoreErrorEventException.test(e))
                throw e;
        }
    }

    /**
     * Process the events—if any—that were produced by the {@link ConsumerNetworkThread network thread}.
     * It is possible that {@link ErrorEvent an error}
     * could occur when processing the events. In such cases, the processor will take a reference to the first
     * error, continue to process the remaining events, and then throw the first error that occurred.
     */
    private boolean processBackgroundEvents() {
        AtomicReference<KafkaException> firstError = new AtomicReference<>();

        LinkedList<BackgroundEvent> events = new LinkedList<>();
        backgroundEventQueue.drainTo(events);
        if (!events.isEmpty()) {
            long startMs = time.milliseconds();
            for (BackgroundEvent event : events) {
                asyncConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
                try {
                    if (event instanceof CompletableEvent)
                        backgroundEventReaper.add((CompletableEvent<?>) event);

                    backgroundEventProcessor.process(event);
                } catch (Throwable t) {
                    KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);

                    if (!firstError.compareAndSet(null, e))
                        log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
                }
            }
            asyncConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
        }

        backgroundEventReaper.reap(time.milliseconds());

        if (firstError.get() != null)
            throw firstError.get();

        return !events.isEmpty();
    }