public void start(ConsumerCallback callback, int pollRate) {
        BlockingBucket bucket = rateLimitBucket(pollRate);
        ConsumerCallback callbackWithRateLimit = (tp, p, st) -> {
            callback.messageReceived(tp, p, st);
            bucket.consume(1);
        };
        CompletableFuture.allOf(
            groups.stream()
                .map(group -> group.start(callbackWithRateLimit))
                .toArray(CompletableFuture[]::new)
        ).join();
    }
Create consumers for the given topics.
Note: the created consumers will start polling immediately.
NOT thread-safe.

@param topics topic names
@param config consumer configuration
@return the number of consumers created
tools/src/main/java/org/apache/kafka/tools/automq/perf/ConsumerService.java