public void close() {
            assert closing;
            if (task != null) {
                try {
                    task.get();
                } catch (Exception e) {
                    LOGGER.error("exception occur while closing consumer", e);
                }
            }
            consumer.close();
        }
Wait for the consumer to finish processing and then close it.
tools/src/main/java/org/apache/kafka/tools/automq/perf/ConsumerService.java