@Override
  public List<String> listTables(ExecutionContext context) throws ExecutorException {
    String address = environmentVariableHandler.getEnvironmentVariable(SAMZA_SQL_SYSTEM_KAFKA_ADDRESS);
    if (address == null || address.isEmpty()) {
      address = DEFAULT_SERVER_ADDRESS;
    }
    try {
      ZkClient zkClient = new ZkClient(address, DEFAULT_ZOOKEEPER_CLIENT_TIMEOUT);
      return zkClient.getChildren(ZOOKEEPER_BROKERS_TOPICS_PATH)
        .stream()
        .map(x -> SAMZA_SYSTEM_KAFKA + "." + x)
        .collect(Collectors.toList());
    } catch (ZkTimeoutException ex) {
      throw new ExecutorException(ex);
    }
  }
Samza implementation of Executor for Samza SQL Shell.
samza-sql-shell/src/main/java/org/apache/samza/sql/client/impl/SamzaExecutor.java