design-proposals/klip-26-java-client-interfaces.md · OSCHINA-MIRROR/mirrors-KSQL

KLIP 26 — Java-клиентские интерфейсы

Автор: Виктория Ся (@vcrfxia) | Целевая версия: 0.10.0; 6.0.0 | Статус: объединён | Обсуждение: GitHub PR

Краткое содержание: KLIP 15 уже привёл аргументы в пользу введения Java-клиента для ksqlDB. Этот KLIP предлагает интерфейсы для клиента.

Мотивация и предыстория

См. KLIP 15.

Что входит в рамки проекта

Java-клиент будет поддерживать следующие операции:

Push-запрос;
Pull-запрос;
Операции DDL SELECT:
- CREATE STREAM;
- CREATE TABLE;
- CREATE STREAM... AS SELECT;
- CREATE TABLE... AS SELECT;
- DROP STREAM;
- DROP TABLE;
- INSERT INTO;
- TERMINATE ;
Операции администрирования SELECT:
- SHOW TOPICS (нерасширенные);
- SHOW STREAMS (нерасширенные);
- SHOW TABLES (нерасширенные);
- SHOW QUERIES (нерасширенные);
Вставка значений, то есть вставка строк в существующий поток;
Завершение push-запроса (через конечную точку /close-query). Цель этого KLIP — достичь соглашения по интерфейсам / публичным API. Детали реализации не рассматриваются.

Что не входит в рамки проекта

Этот KLIP не охватывает поддержку Java-клиентом следующих функций:

DESCRIBE <stream/table>, DESCRIBE FUNCTION;
EXPLAIN ;
PRINT ;
SHOW TOPICS EXTENDED, SHOW <STREAMS/TABLES> EXTENDED, SHOW QUERIES EXTENDED;
CREATE CONNECTOR, DROP CONNECTOR, SHOW CONNECTORS, DESCRIBE CONNECTOR;
CREATE TYPE, DROP TYPE, SHOW TYPES;
SHOW FUNCTIONS, SHOW PROPERTIES;
RUN SCRIPT;
Использование других конечных точек (info, healthcheck, terminate cluster, status и т. д.). Мы всегда можем добавить поддержку этих операций в будущем, если потребуется. Как и выше, детали реализации выходят за рамки, поскольку цель этого KLIP состоит в том, чтобы достичь согласия по интерфейсам / публичным API.

Значение/возврат

См. KLIP 15.

Публичные API

В следующих подразделах описаны методы интерфейса Client:

public interface Client {
    ...
}

Конструкторы

static Client create(ClientOptions clientOptions) {
  return new ClientImpl(clientOptions);
}

static Client create(ClientOptions clientOptions, Vertx vertx) {
  return new ClientImpl(clientOptions, vertx);
}

Java-клиент будет реализован как Vert.x HttpClient. Мы предоставляем конструктор, который позволяет пользователям предоставлять свой собственный экземпляр Vertx, чтобы при желании воспользоваться общим пулом соединений и другими свойствами.

ClientOptions изначально будет выглядеть следующим образом:

public interface ClientOptions {

  ClientOptions setHost(String host);

  ClientOptions setPort(int port);

  ClientOptions setUseTls(boolean useTls);
  
  ClientOptions setVerifyHost(boolean verifyHost);
  
  ClientOptions setUseAlpn(boolean useAlpn);

  ClientOptions setTrustStore(String trustStorePath);

  ClientOptions setTrustStorePassword(String trustStorePassword);

  ClientOptions setKeyStore(String keyStorePath);

  ClientOptions setKeyStorePassword(String keyStorePassword);

  ClientOptions setBasicAuthCredentials(String username, String password);
  
  ClientOptions setExecuteQueryMaxResultRows(int maxRows);

  String getHost();

  int getPort();

  boolean isUseTls();
  
  boolean isVerifyHost();
  
  boolean isUseAlpn();

  boolean isUseBasicAuth();

  String getTrustStore();

  String getTrustStorePassword();

  String getKeyStore();

  String getKeyStorePassword();

  String getBasicAuthUsername();

  String getBasicAuthPassword();

  ClientOptions copy();

  static ClientOptions create() {
    return new ClientOptionsImpl();
  }
}

Мы всегда сможем добавить дополнительные параметры конфигурации позже. Также мы можем захотеть предоставить пользователям доступ к Vert.x HttpClientOptions для опытных пользователей, которые хотят предоставить собственные настройки. Выполняет запрос (push или pull) и возвращает результаты по одной строке за раз.

Если от сервера получен ответ, отличный от 200, то CompletableFuture будет неудачным.
По умолчанию запросы push, отправленные через этот метод, возвращают результаты, начиная с начала потока или таблицы. Чтобы переопределить это поведение, используйте метод {@link #streamQuery(String, Map)}, чтобы передать в запросе свойство {@code auto.offset.reset} со значением, установленным в {@code latest}.

@param sql оператор запроса для выполнения @return будущее, которое завершается после получения ответа сервера и содержит результат запроса в случае успеха

Выполняет запрос (push или pull) и возвращает результаты по одной строке за раз.

Если от сервера получен ответ, отличный от 200, то CompletableFuture будет неудачным.
Запросы push, отправленные с помощью этого метода, по умолчанию возвращают результаты, начинающиеся с начала потока или таблицы. Для переопределения этого поведения передайте в запросе свойство {@code auto.offset.reset}, установив значение {@code latest}.

@param sql оператор запроса для выполнения @param properties свойства запроса @return будущее, которое завершается после получения ответа сервера и содержит результат запроса в случае успеха

где StreamedQueryResult выглядит следующим образом:

Результат запроса (push или pull), передаваемый по потоку по одной строке за раз. Записи могут быть использованы либо путём подписки на издателя, либо путём опроса (блокировки) по одной записи за раз. Эти два способа потребления являются взаимоисключающими; может использоваться только один метод (для каждого StreamedQueryResult).

Метод {@code subscribe()} не может быть вызван, если {@link #isFailed} истинно.

public interface StreamedQueryResult extends Publisher {

List columnNames();

List columnTypes();

/**

Возвращает идентификатор основного запроса, если запрос является запросом push. В противном случае возвращает null.
@return идентификатор запроса */ String queryID();

/**

Возвращает следующую строку. Блокируется до тех пор, пока она не станет доступна или основной запрос не будет завершён (либо корректно, либо из-за ошибки).
@return строка или null, если запрос был завершён. */ Row poll();

/**

Возвращает следующую строку. Блокируется до тех пор, пока она не станет доступна, не истечёт указанный тайм-аут или основной запрос не будет завершён (либо корректно, либо из-за ошибки).
@param timeout время ожидания строки. Неположительное значение приведёт к блокировке этого метода до получения строки или завершения запроса.
@return строка, или null, если истекло время ожидания или запрос был завершён. */ Row poll(Duration timeout);

/**

Определяет, завершён ли {@code StreamedQueryResult}.
{@code StreamedQueryResult} считается завершённым, если HTTP-соединение, связанное с этим запросом, было корректно закрыто. После завершения {@code StreamedQueryResult} продолжит доставлять любые оставшиеся строки, а затем вызовет {@code onComplete()} у подписчика, если таковой имеется.
@return завершённость {@code StreamedQueryResult}. */ boolean isComplete();

/**

Определяет, завершился ли {@code StreamedQueryResult} неудачно.
Если {@code StreamedQueryResult} завершился неудачно, то от сервера была получена ошибка. После неудачи {@code onError()} вызывается у подписчика, если он есть, все существующие вызовы {@code poll()} вернут null, а новые вызовы {@link #poll} и {@code subscribe()} будут отклонены.
@return неудачное завершение {@code StreamedQueryResult}. */ boolean isFailed(); }

Обратите внимание, что StreamedQueryResult — это издатель Reactive Streams, поэтому пользователи могут передавать результаты. Пользователи также могут вызывать poll(), чтобы... Вот перевод текста на русский язык:

  /**
   * Возвращает значение для определённого столбца {@code Row} как long.
   */
  Long getLong(String columnName);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде double.
   *
   * @param columnIndex индекс столбца (отсчёт с 1)
   * @return значение столбца
   * @throws ClassCastException если значение столбца не является числом {@code Number}
   * @throws IndexOutOfBoundsException если индекс недопустим
   */
  Double getDouble(int columnIndex);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде double.
   *
   * @param columnName имя столбца
   * @return значение столбца
   * @throws ClassCastException если значение столбца не является числом {@code Number}
   * @throws IllegalArgumentException если имя столбца недопустимо
   */
  Double getDouble(String columnName);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде boolean.
   *
   * @param columnIndex индекс столбца (отсчёт с 1)
   * @return значение столбца
   * @throws ClassCastException если значение столбца не является логическим значением
   * @throws IndexOutOfBoundsException если индекс недопустим
   */
  Boolean getBoolean(int columnIndex);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде boolean.
   *
   * @param columnName имя столбца
   * @return значение столбца
   * @throws ClassCastException если значение столбца не является логическим значением
   * @throws IllegalArgumentException если имя столбца недопустимо
   */
  Boolean getBoolean(String columnName);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде BigDecimal.
   *
   * @param columnIndex индекс столбца (отсчёт с 1)
   * @return значение столбца
   * @throws ClassCastException если значение столбца не является числом {@code Number}
   * @throws IndexOutOfBoundsException если индекс недопустим
   */
  BigDecimal getDecimal(int columnIndex);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде BigDecimal.
   *
   * @param columnName имя столбца
   * @return значение столбца
   * @throws ClassCastException если значение столбца не является числом {@code Number}
   * @throws IllegalArgumentException если имя столбца недопустимо
   */
  BigDecimal getDecimal(String columnName);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде KsqlObject.
   * Полезно для типов столбцов {@code MAP} и {@code STRUCT}.
   *
   * @param columnIndex индекс столбца (отсчёт с 1)
   * @return значение столбца
   * @throws ClassCastException если значение столбца нельзя преобразовать в карту
   * @throws IndexOutOfBoundsException если индекс недопустим
   */
  KsqlObject getKsqlObject(int columnIndex);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде KsqlObject.
   * Полезно для типов столбцов {@code MAP} и {@code STRUCT}.
   *
   * @param columnName имя столбца
   * @return значение столбца
   * @throws ClassCastException если значение столбца нельзя преобразовать в карту
   * @throws IllegalArgumentException если имя столбца недопустимо
   */
  KsqlObject getKsqlObject(String columnName);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде KsqlArray.
   * Полезно для типа столбца {@code ARRAY}.
   *
   * @param columnIndex индекс столбца (отсчёт с 1)
   * @return значение столбца
   * @throws ClassCastException если значение столбца нельзя преобразовать в список
   * @throws IndexOutOfBoundsException если индекс недопустим
   */
  KsqlArray getKsqlArray(int columnIndex);

  /**
   * Возвращает значение определённого столбца {@code Row} в виде KsqlArray.
   * Полезно для типа столбца {@code ARRAY}.
   *
   * @param columnName имя столбца
   * @return значение столбца
   * @throws ClassCastException если значение столбца нельзя преобразовать в список
   * @throws IllegalArgumentException если имя столбца недопустимо
   */
  KsqlaArray getKsqlArray(String columnName);
}
``` **Сериализация в и из JSON**) — это то, что мы можем использовать в будущем.

Мы решили ввести новые типы вместо того, чтобы напрямую использовать `List<Object>` и `Map<String, Object>`, чтобы обеспечить большую гибкость для развития этих API в будущем.
(Мне не нравятся названия `KsqlArray` и `KsqlObject`, но я не смог придумать ничего лучше. Буду признателен за предложения!)

Методы, предоставляемые каждым типом, следующие:

/**

Представление массива значений. */ public KsqlArray {

/**

Создаёт пустой экземпляр. */ public KsqlArray() { delegate = new JsonArray(); }

/**

Создает экземпляр с указанными значениями.
@param list значения */ public KsqlArray(final List<?> list) { delegate = new JsonArray(list); }

/**

Возвращает размер (количество значений) массива.
@return размер */ public int size() { return delegate.size(); }

/**

Проверяет, пуст ли массив.
@return является ли массив пустым */ public boolean isEmpty() { return delegate.isEmpty(); }

/**

Возвращает значения массива в виде списка.
@return список значений */ public List<?> getList() { return delegate.getList(); }

/**

Возвращает итератор по значениям массива.
@return итератор */ public Iterator iterator() { return delegate.iterator(); }

/**

Возвращает поток значений массива.
@return поток */ public java.util.stream.Stream stream() { return delegate.stream(); }

/**

Возвращает значение по указанному индексу как объект.
@param pos индекс
@return значение */ public Object getValue(final int pos) { return delegate.getValue(pos); }

/**

Возвращает значение по указанному индексу в виде строки.
@param pos индекс
@return значение
@throws ClassCastException если значение не является строкой
@throws IndexOutOfBoundsException если индекс недействителен */ public String getString(final int pos) { return delegate.getString(pos); }

/**

Возвращает значение по указанному индексу в виде целого числа.
@param pos индекс
@return значение
@throws ClassCastException если значение не относится к типу {@code Number}
@throws IndexOutOfBoundsException если индекс недействителен */ public Integer getInteger(final int pos) { return delegate.getInteger(pos); }

/**

Возвращает значение по указанному индексу в виде длинного числа.
@param pos индекс
@return значение
@throws ClassCastException если значение не относится к типу {@code Number}
@throws IndexOutOfBoundsException если индекс недействителен */ public Long getLong(final int pos) { return delegate.getLong(pos); }

/**

Возвращает значение по указанному индексу в виде двойного числа.
@param pos индекс
@return значение
@throws ClassCastException если значение не относится к типу {@code Number}
@throws IndexOutOfBoundsException если индекс недействителен */ public Double getDouble(final int pos) { return delegate.getDouble(pos); }

/**

Возвращает значение по указанному индексу в виде логического значения.
@param pos индекс
@return значение
@throws ClassCastException если значение не является логическим
@throws IndexOutOfBoundsException если индекс недействителен */ public Boolean getBoolean(final int pos) { return delegate.getBoolean(pos); }

/**

Возвращает значение по указанному индексу в виде BigDecimal.
@param pos индекс
@return значение
@throws ClassCastException если значение не относится к типу {@code Number}
@throws IndexOutOfBoundsException если индекс недействителен */ public BigDecimal getDecimal(final int pos) { return new BigDecimal(getValue(pos).toString()); }

/**

Возвращает значение по указанному индексу в виде KsqlArray.
@param pos индекс
@return значение
@throws ClassCastException если значение нельзя преобразовать в список
@throws IndexOutOfBoundsException */ Текст 1:
@return the JSON string

public String toJsonString() { return delegate.toString(); }

/** Возвращает строку JSON.

@return строка JSON

@Override

public String toString() { return toJsonString(); } }*

Текст 2:

/** Представление карты строковых ключей и значений. Полезно для представления объекта JSON.

public KsqlObject {

public KsqlObject() { delegate = new JsonObject(); }**

public KsqlObject(final Map<String, Object> map) { delegate = new JsonObject(map); }**

KsqlObject(final JsonObject jsonObject) { delegate = Objects.requireNonNull(jsonObject); }**

public boolean containsKey(final String key) { return delegate.containsKey(key); }**

public Set fieldNames() { return delegate.fieldNames(); }**

public int size() { return delegate.size(); }**

public boolean isEmpty() { return delegate.isEmpty(); }**

public Map<String, Object> getMap() { return delegate.getMap(); }**

public Iterator<Entry<String,Object>> iterator() { return delegate.iterator(); }**

public java.util.stream.Stream<Map.Entry<String,Object>> stream() { return delegate.stream(); }**

public Object getValue(final String key) { return delegate.getValue(key); }**

public String getString(final String key) { return delegate.getString(key); }**

public Integer getInteger(final String key) { return delegate.getInteger(key); }**

public Long getLong(final String key) { return delegate.getLong(key); }**

public Double getDouble(final String key) { return delegate.getDouble(key); }**

public Boolean getBoolean(final String key) { return delegate.getBoolean(key); }**

public void put(final String key, final Object value) { Objects.requireNonNull(key); if (value == null) { throw new NullPointerException("value"); }** delegate.put(key, value); }**

public void remove(final String key) { Objects.requireNonNull(key); delegate.remove(key); }**

public static class Builder { private final Map<String, Object> entries = new HashMap<>();

 ***public Builder put(final String key, final Object value) {***
 *entries.put(key, value);*
 *return this;*

}**

public Builder remove(final String key) { entries.remove(key); return this; }**

public KsqlObject build() { return new KsqlObject(entries); }}}*

В запросе представлен исходный код на языке Java, который описывает класс KsqlObject, представляющий собой карту строковых ключей и значений, полезных для представления объектов JSON. Класс содержит методы для работы с картой, такие как добавление, удаление и получение значений по ключу. Также в коде представлен конструктор класса и статический внутренний класс Builder, который позволяет удобно создавать экземпляры KsqlObject с заданными значениями. Как BigDecimal. Возвращает null, если:

ключ отсутствует;
@param key ключ;
@return значение;
@throws ClassCastException если значение не является числом.

public BigDecimal getDecimal(final String key) { return new BigDecimal(getValue(key).toString()); }

Возвращает значение, связанное с указанным ключом как KsqlArray. Возвращает null, если ключ отсутствует.

@param key ключ;
@return значение;
@throws ClassCastException если значение нельзя преобразовать в список.

public KsqlArray getKsqlArray(final String key) { return new KsqlArray(delegate.getJsonArray(key)); }

Возвращает значение, связанное с указанным ключом, как KsqlObject. Возвращает null, если ключ отсутствует.

@param key ключ;
- @return значение;
@throws ClassCastException если значение невозможно преобразовать в карту.

public KsqlObject getKsqlObject(final String key) { return new Ksqlobject(delegate.getJsonObject(key)); }

Удаляет значение, связанное с заданным ключом.

@param key ключ;
@return удалённое значение или null, если ключа не было.

public Object remove(final String key) { return delegate.remove(key); }