Skip to content


Creating Instance

Cli takes the following to be created:

  • streamedQueryRowLimit (based on query-row-limit option)
  • streamedQueryTimeoutMs (based on query-timeout option)
  • KsqlRestClient
  • Console

Cli is created using build utility.


Cli is given a KsqlRestClient when created.

The KsqlRestClient is used when:


Cli creates STATEMENT_HANDLERS internal map when created to handle KSQL statements.

Statement Handler
QueryStatementContext handleQuery
PrintTopicContext handlePrintedTopic
SetPropertyContext setPropertyFromCtxt
UnsetPropertyContext unsetPropertyFromCtxt
DefineVariableContext defineVariableFromCtxt
UndefineVariableContext undefineVariableFromCtxt
ListVariablesContext listVariablesFromCtxt
CreateConnectorContext handleConnectorRequest
DropConnectorContext handleConnectorRequest
DescribeConnectorContext handleConnectorRequest
ListConnectorsContext handleConnectorRequest
ListConnectorPluginsContext handleConnectorRequest


STATEMENT_HANDLERS is a static final value so it is initialized when Cli is loaded by the JVM.

Building Cli Instance

Cli build(
  Long streamedQueryRowLimit,
  Long streamedQueryTimeoutMs,
  OutputFormat outputFormat,
  KsqlRestClient restClient

build builds a Console (for the OutputFormat) to create a Cli.

build is used when:

  • Ksql is requested to run


void runCommand(
  String command)

runCommand handleLine.

runCommand is used when:

  • Ksql is requested to run


void runInteractively()

runInteractively displayWelcomeMessage.

runInteractively validates the KsqlRestClient.

runInteractively handleLine until stopped (by a user).

runInteractively is used when:

  • Ksql is requested to run


void handleLine(
  String line)

handleLine removes any leading and trailing spaces from the given line and handleStatements.

handleLine simply returns back when the given line is empty after trimming.

handleLine is used when:

Handling KSQL Statements

void handleStatements(
  String line)

handleStatements requests the DefaultKsqlParser to parse the given line (into ParsedStatements).


There could be one or more ParsedStatements in the given line.

For every ParsedStatement, handleStatements substituteVariables and...FIXME

handleStatements validates the statements.

handleStatements executes the statements:

  1. substituteVariables (with isSandbox flag disabled)
  2. Looks up the handler (in the STATEMENT_HANDLERS) to handle the statement


ParsedStatement substituteVariables(
  ParsedStatement statement)



boolean isVariableSubstitutionEnabled()



void makeKsqlRequest(
  String statements)

makeKsqlRequest is part of the KsqlRequestExecutor abstraction.

makeKsqlRequest makes a ksql request with the statements (and the KsqlRestClient) and printKsqlResponse to the console.

makeKsqlRequest (private)

RestResponse<R> makeKsqlRequest(
  final String ksql,
  final BiFunction<String, Long, RestResponse<R>> requestIssuer)

makeKsqlRequest executes the requestIssuer binary function (that uses the KsqlRestClient) with the given ksql (and commandSequenceNumberToWaitFor if configured).

makeKsqlRequest retires execution of failed statements 10 times.

makeKsqlRequest is used when:


void handleQuery(
  String statement,
  SqlBaseParser.QueryStatementContext query)

handleQuery makeKsqlRequest (to /query endpoint).

In the end, handleQuery prints an error message (if the request was unsuccessful) or prints out the response until CTRL-C or the query terminated.

handleQuery is used when: