public static StreamedRow toRowFromDelimited(final Buffer buff) {
    try {
      final QueryResponseMetadata metadata = deserialize(buff, QueryResponseMetadata.class);
      return StreamedRow.header(new QueryId(Strings.nullToEmpty(metadata.queryId)),
        createSchema(metadata));
    } catch (KsqlRestClientException e) {
      // Not a {@link QueryResponseMetadata}
    }
    try {
      final KsqlErrorMessage error = deserialize(buff, KsqlErrorMessage.class);
      return StreamedRow.error(new RuntimeException(error.getMessage()), error.getErrorCode());
    } catch (KsqlRestClientException e) {
      // Not a {@link KsqlErrorMessage}
    }
    try {
      final PushContinuationToken continuationToken
          = deserialize(buff, PushContinuationToken.class);
      return StreamedRow.continuationToken(continuationToken);
    } catch (KsqlRestClientException e) {
      // Not a {@link KsqlErrorMessage}
    }
    try {
      final List<?> row = deserialize(buff, List.class);
      return StreamedRow.pushRow(GenericRow.fromList(row));
    } catch (KsqlRestClientException e) {
      // Not a {@link List}
    }
    throw new IllegalStateException("Couldn't parse message: " + buff.toString());
  }
@Test
  public void shouldParseHeader() {
    // When:
    final StreamedRow row = KsqlTargetUtil.toRowFromDelimited(Buffer.buffer(
        "{\"queryId\": \"query_id_10\", "
            + "\"columnNames\":[\"col1\",\"col2\"], "
            + "\"columnTypes\":[\"BIGINT\",\"DOUBLE\"]}"));

    // Then:
    assertThat(row.getHeader().isPresent(), is(true));
    assertThat(row.getHeader().get().getQueryId().toString(), is("query_id_10"));
    assertThat(row.getHeader().get().getSchema().key(), is(Collections.emptyList()));
    assertThat(row.getHeader().get().getSchema().value().size(), is(2));
    assertThat(row.getHeader().get().getSchema().value().get(0),
        is (Column.of(ColumnName.of("col1"), SqlTypes.BIGINT, Namespace.VALUE, 0)));
    assertThat(row.getHeader().get().getSchema().value().get(1),
        is (Column.of(ColumnName.of("col2"), SqlTypes.DOUBLE, Namespace.VALUE, 1)));
  }