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

Code before:
    @Test
    public void testRecordBackgroundEventQueueSizeAndBackgroundEventQueueTime() {
        consumer = newConsumer(
                mock(FetchBuffer.class),
                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);

        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;
    }

Review comment (bug):
Interesting, trying to understand why we needed this, I realized that it was probably to make sure there is a non-null `groupMetadata`, so you can ensure that the test pass because it hits they newly added null check on `appEventHandler`, and not because of the existing early return on groupMetadata.isEmpty, correct? But then I realized that the `groupMetadata` creation happens after the metrics anyways, so it will end up always being null for this test anyways, right? (meaning I expect this test would pass, no noisy NPE even without this PR)

That being said, I think the sanity checks make sense, and the test will surely help catch any regression (a simple change or order in the actions of the constructor would make the noisy NPEs appear I expect).

So no changes I would say, just thinking out loud here and sharing for the record.

Code after:
    @Test
    public void testRecordBackgroundEventQueueSizeAndBackgroundEventQueueTime() {
        consumer = newConsumer(
                mock(FetchBuffer.class),
                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;