@Override
        public void complete(LineReader reader, ParsedLine line, List<Candidate> candidates) {
            if (line.words().isEmpty()) {
                CommandUtils.completeCommand("", candidates);
            } else if (line.words().size() == 1) {
                CommandUtils.completeCommand(line.words().get(0), candidates);
            } else {
                Iterator<String> iter = line.words().iterator();
                String command = iter.next();
                List<String> nextWords = new ArrayList<>();
                while (iter.hasNext()) {
                    nextWords.add(iter.next());
                }
                Commands.Type type = Commands.TYPES.get(command);
                if (type == null) {
                    return;
                }
                try {
                    type.completeNext(state, nextWords, candidates);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
Handles running the Kafka metadata shell in interactive mode, where we accept input in real time.
shell/src/main/java/org/apache/kafka/shell/InteractiveShell.java