chia7712 commented on code in PR #15821:
URL: https://github.com/apache/kafka/pull/15821#discussion_r1600210615
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommandTestUtils.java:
##########
@@ -93,6 +95,27 @@ static <T> AutoCloseable buildConsumers(int
numberOfConsumers,
}
}
+ static <T> AutoCloseable buildConsumers(int numberOfConsumers,
Review Comment:
Please DON'T generate a bunch of duplicate code...
```java
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
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
