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

Code before:
                mock(ConsumerRebalanceListenerInvoker.class),
                mock(SubscriptionState.class));
        Metrics metrics = consumer.metricsRegistry();
        AsyncConsumerMetrics asyncConsumerMetrics = consumer.asyncConsumerMetrics();

        ConsumerRebalanceListenerCallbackNeededEvent event = new ConsumerRebalanceListenerCallbackNeededEvent(ON_PARTITIONS_REVOKED, Collections.emptySortedSet());
        event.setEnqueuedMs(time.milliseconds());
        backgroundEventQueue.add(event);
        asyncConsumerMetrics.recordBackgroundEventQueueSize(1);

        time.sleep(10);
        consumer.processBackgroundEvents();
        assertEquals(0, (double) metrics.metric(metrics.metricName("background-event-queue-size", CONSUMER_METRIC_GROUP)).metricValue());
        assertEquals(10, (double) metrics.metric(metrics.metricName("background-event-queue-time-avg", CONSUMER_METRIC_GROUP)).metricValue());
        assertEquals(10, (double) metrics.metric(metrics.metricName("background-event-queue-time-max", CONSUMER_METRIC_GROUP)).metricValue());
    }

    private Map<TopicPartition, OffsetAndMetadata> mockTopicPartitionOffset() {
        final TopicPartition t0 = new TopicPartition("t0", 2);
        final TopicPartition t1 = new TopicPartition("t0", 3);
        Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsets = new HashMap<>();
        topicPartitionOffsets.put(t0, new OffsetAndMetadata(10L));
        topicPartitionOffsets.put(t1, new OffsetAndMetadata(20L));
        return topicPartitionOffsets;
    }

    private Map<TopicPartition, OffsetAndTimestampInternal> mockOffsetAndTimestamp() {
        final TopicPartition t0 = new TopicPartition("t0", 2);
        final TopicPartition t1 = new TopicPartition("t0", 3);
        Map<TopicPartition, OffsetAndTimestampInternal> offsetAndTimestamp = new HashMap<>();
        offsetAndTimestamp.put(t0, new OffsetAndTimestampInternal(5L, 1L, Optional.empty()));
        offsetAndTimestamp.put(t1, new OffsetAndTimestampInternal(6L, 3L, Optional.empty()));
        return offsetAndTimestamp;
    }

    private Map<TopicPartition, Long> mockTimestampToSearch() {
        final TopicPartition t0 = new TopicPartition("t0", 2);
        final TopicPartition t1 = new TopicPartition("t0", 3);
        Map<TopicPartition, Long> timestampToSearch = new HashMap<>();
        timestampToSearch.put(t0, 1L);
        timestampToSearch.put(t1, 2L);
        return timestampToSearch;
    }

    private void completeCommitAsyncApplicationEventExceptionally(Exception ex) {
        doAnswer(invocation -> {
            AsyncCommitEvent event = invocation.getArgument(0);
            event.markOffsetsReady();
            event.future().completeExceptionally(ex);
            return null;
        }).when(applicationEventHandler).add(ArgumentMatchers.isA(AsyncCommitEvent.class));

Review comment (bug):
It seems that code was copied from `ShareConsumerImplTest#testFailConstructor` with some improvements -- for example, the `appender` now gets closed :)

@gensericghiro Do you have a free cycle to file a MINOR patch to fix it in `ShareConsumerImplTest#testFailConstructor`?

Code after:
                mock(ConsumerInterceptors.class),
                mock(ConsumerRebalanceListenerInvoker.class),
                mock(SubscriptionState.class));
        Metrics metrics = consumer.metricsRegistry();
        AsyncConsumerMetrics asyncConsumerMetrics = consumer.asyncConsumerMetrics();

        ConsumerRebalanceListenerCallbackNeededEvent event = new ConsumerRebalanceListenerCallbackNeededEvent(ON_PARTITIONS_REVOKED, Collections.emptySortedSet());
        event.setEnqueuedMs(time.milliseconds());
        backgroundEventQueue.add(event);
        asyncConsumerMetrics.recordBackgroundEventQueueSize(1);

        time.sleep(10);
        consumer.processBackgroundEvents();
        assertEquals(0, (double) metrics.metric(metrics.metricName("background-event-queue-size", CONSUMER_METRIC_GROUP)).metricValue());
        assertEquals(10, (double) metrics.metric(metrics.metricName("background-event-queue-time-avg", CONSUMER_METRIC_GROUP)).metricValue());
        assertEquals(10, (double) metrics.metric(metrics.metricName("background-event-queue-time-max", CONSUMER_METRIC_GROUP)).metricValue());
    }

    @Test
    public void testFailConstructor() {
        final Properties props = requiredConsumerConfig();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
        props.put(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG, "an.invalid.class");
        final ConsumerConfig config = new ConsumerConfig(props);

        try (LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
            KafkaException ce = assertThrows(
                KafkaException.class,
                () -> newConsumer(config));
            assertTrue(ce.getMessage().contains("Failed to construct kafka consumer"), "Unexpected exception message: " + ce.getMessage());
            assertTrue(ce.getCause().getMessage().contains("Class an.invalid.class cannot be found"), "Unexpected cause: " + ce.getCause());

            boolean npeLogged = appender.getEvents().stream()
                .flatMap(event -> event.getThrowableInfo().stream())
                .anyMatch(str -> str.contains("NullPointerException"));

            assertFalse(npeLogged, "Unexpected NullPointerException during consumer construction");
        }
    }

    private Map<TopicPartition, OffsetAndMetadata> mockTopicPartitionOffset() {
        final TopicPartition t0 = new TopicPartition("t0", 2);
        final TopicPartition t1 = new TopicPartition("t0", 3);
        Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsets = new HashMap<>();
        topicPartitionOffsets.put(t0, new OffsetAndMetadata(10L));
        topicPartitionOffsets.put(t1, new OffsetAndMetadata(20L));
        return topicPartitionOffsets;
    }

    private Map<TopicPartition, OffsetAndTimestampInternal> mockOffsetAndTimestamp() {
        final TopicPartition t0 = new TopicPartition("t0", 2);