MINOR: migrate ListConsumerGroupTest to use ClusterTestExtensions

[FrankYang0529]

By using ClusterTestExtensions, ListConsumerGroupTest get get away from KafkaServerTestHarness dependency.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[FrankYang0529]

Rely on #15766.

[FrankYang0529]

Hi @chia7712, I rebase latest trunk branch, so we have ConsumerGroupCommandTestUtils now. The only remaining part is SimpleConsumerGroupExecutor. This is used by ListConsumerGroupTest and DescribeConsumerGroupTest. I think we can create a new class SimpleConsumerGroupExecutorTestUtils for it. WDYT? Thank you.

kafka/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandTest.java

Lines 327 to 335 in 21bf715

	class SimpleConsumerGroupExecutor extends AbstractConsumerGroupExecutor {
	public SimpleConsumerGroupExecutor(String broker, String groupId, Collection<TopicPartition> partitions) {
	super(1);
	SimpleConsumerRunnable th = new SimpleConsumerRunnable(broker, groupId, partitions);
	th.configure();
	submit(th);
	}
	}

[chia7712]

This is used by ListConsumerGroupTest and DescribeConsumerGroupTest. I think we can create a new class SimpleConsumerGroupExecutorTestUtils for it. WDYT? Thank you.

That is addressed already. see
https://github.com/apache/kafka/blob/trunk/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandTestUtils.java#L92

[FrankYang0529]

This is used by ListConsumerGroupTest and DescribeConsumerGroupTest. I think we can create a new class SimpleConsumerGroupExecutorTestUtils for it. WDYT? Thank you.

That is addressed already. see https://github.com/apache/kafka/blob/trunk/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandTestUtils.java#L92

Thanks. The SimpleConsumerGroupExecutor subscribe topic partitions. I have updated ConsumerGroupCommandTestUtils to support it.

kafka/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandTest.java

Lines 274 to 287 in f4fdaa7

	class SimpleConsumerRunnable extends AbstractConsumerRunnable {
	final Collection<TopicPartition> partitions;
	public SimpleConsumerRunnable(String broker, String groupId, Collection<TopicPartition> partitions) {
	super(broker, groupId, Optional.empty(), false);
	this.partitions = partitions;
	}
	@Override
	void subscribe() {
	consumer.assign(partitions);
	}
	}

[chia7712]

Could we have two buildConsumers to deal with "assign"/"subscribe" individually?

[FrankYang0529]

Thank you. I update to define two buildConsumers. May you help me take a look? If we're good with current design. I will update to #15908.

[chia7712]

@FrankYang0529 please rebase code to include 334d5d5

[chia7712]

Please DON'T generate a bunch of duplicate code...

    static <T> AutoCloseable buildConsumers(int numberOfConsumers,
                                            boolean syncCommit,
                                            Set<TopicPartition> partitions,
                                            Supplier<KafkaConsumer<T, T>> consumerSupplier) {
        return buildConsumers(numberOfConsumers, syncCommit, consumerSupplier,
            consumer -> consumer.assign(partitions));
    }

    static <T> AutoCloseable buildConsumers(int numberOfConsumers,
                                            boolean syncCommit,
                                            String topic,
                                            Supplier<KafkaConsumer<T, T>> consumerSupplier) {
        return buildConsumers(numberOfConsumers, syncCommit, consumerSupplier,
            consumer -> consumer.subscribe(Collections.singleton(topic)));
    }

    private static <T> AutoCloseable buildConsumers(int numberOfConsumers,
                                                    boolean syncCommit,
                                                    Supplier<KafkaConsumer<T, T>> consumerSupplier,
                                                    Consumer<KafkaConsumer<T, T>> setPartitions) {
        List<KafkaConsumer<T, T>> consumers = new ArrayList<>(numberOfConsumers);
        ExecutorService executor = Executors.newFixedThreadPool(numberOfConsumers);
        AtomicBoolean closed = new AtomicBoolean(false);
        final AutoCloseable closeable = () -> releaseConsumers(closed, consumers, executor);
        try {
            for (int i = 0; i < numberOfConsumers; i++) {
                KafkaConsumer<T, T> consumer = consumerSupplier.get();
                consumers.add(consumer);
                executor.execute(() -> initConsumer(syncCommit, () -> {
                    setPartitions.accept(consumer);
                    return consumer;
                }, closed));
            }
            return closeable;
        } catch (Throwable e) {
            Utils.closeQuietly(closeable, "Release Consumer");
            throw e;
        }
    }

    private static <T> void releaseConsumers(AtomicBoolean closed, List<KafkaConsumer<T, T>> consumers, ExecutorService executor) throws InterruptedException {
        closed.set(true);
        consumers.forEach(KafkaConsumer::wakeup);
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
    }

    private static <T> void initConsumer(boolean syncCommit,
                                         Supplier<KafkaConsumer<T, T>> consumerSupplier,
                                         AtomicBoolean closed) {
        try (KafkaConsumer<T, T> kafkaConsumer = consumerSupplier.get()) {
            while (!closed.get()) {
                kafkaConsumer.poll(Duration.ofMillis(Long.MAX_VALUE));
                if (syncCommit)
                    kafkaConsumer.commitSync();
            }
        } catch (WakeupException e) {
            // OK
        }
    }