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

Code before:
    }

    @ParameterizedTest
    @EnumSource(GroupProtocol.class)
    public void testCloseShouldBeIdempotent(GroupProtocol groupProtocol) {
        ConsumerMetadata metadata = createMetadata(subscription);
        MockClient client = spy(new MockClient(time, metadata));
        initMetadata(client, Map.of(topic, 1));

        consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false, groupInstanceId);

        consumer.close(CloseOptions.timeout(Duration.ZERO));
        consumer.close(CloseOptions.timeout(Duration.ZERO));

        // verify that the call is idempotent by checking that the network client is only closed once.
        verify(client).close();
    }

    @SuppressWarnings("resource")
    @ParameterizedTest
    @EnumSource(GroupProtocol.class)
    public void testOperationsBySubscribingConsumerWithDefaultGroupId(GroupProtocol groupProtocol) {
        try {
            newConsumer(groupProtocol, null, Optional.of(Boolean.TRUE));
            fail("Expected an InvalidConfigurationException");
        } catch (InvalidConfigurationException e) {
            // OK, expected
        }

        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            assertThrows(InvalidGroupIdException.class, () -> consumer.subscribe(Set.of(topic)));
        }

        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            assertThrows(InvalidGroupIdException.class, () -> consumer.committed(Set.of(tp0)).get(tp0));
        }

        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            assertThrows(InvalidGroupIdException.class, consumer::commitAsync);
        }

        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            assertThrows(InvalidGroupIdException.class, consumer::commitSync);
        }
    }

    @ParameterizedTest
    @EnumSource(GroupProtocol.class)
    public void testOperationsByAssigningConsumerWithDefaultGroupId(GroupProtocol groupProtocol) {
        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            consumer.assign(Set.of(tp0));

Review comment (suggestion):
This is a bad pattern: `assertThrow` should be used instead.

Code after:
    }

    @ParameterizedTest
    @EnumSource(GroupProtocol.class)
    public void testCloseShouldBeIdempotent(GroupProtocol groupProtocol) {
        ConsumerMetadata metadata = createMetadata(subscription);
        MockClient client = spy(new MockClient(time, metadata));
        initMetadata(client, Map.of(topic, 1));

        consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false, groupInstanceId);

        consumer.close(CloseOptions.timeout(Duration.ZERO));
        consumer.close(CloseOptions.timeout(Duration.ZERO));

        // verify that the call is idempotent by checking that the network client is only closed once.
        verify(client).close();
    }

    @SuppressWarnings("resource")
    @ParameterizedTest
    @EnumSource(GroupProtocol.class)
    public void testOperationsBySubscribingConsumerWithDefaultGroupId(GroupProtocol groupProtocol) {
        try {
            newConsumer(groupProtocol, null, Optional.of(Boolean.TRUE));
            fail("Expected an InvalidConfigurationException");
        } catch (InvalidConfigurationException swallow) {
            // OK, expected
        }

        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            assertThrows(InvalidGroupIdException.class, () -> consumer.subscribe(Set.of(topic)));
        }

        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            assertThrows(InvalidGroupIdException.class, () -> consumer.committed(Set.of(tp0)).get(tp0));
        }

        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            assertThrows(InvalidGroupIdException.class, consumer::commitAsync);
        }

        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            assertThrows(InvalidGroupIdException.class, consumer::commitSync);
        }
    }

    @ParameterizedTest
    @EnumSource(GroupProtocol.class)
    public void testOperationsByAssigningConsumerWithDefaultGroupId(GroupProtocol groupProtocol) {
        try (KafkaConsumer<byte[], byte[]> consumer = newConsumer(groupProtocol, null)) {
            consumer.assign(Set.of(tp0));