public static List<StreamedRow> toRows(
      final Buffer buff,
      final Function<StreamedRow, StreamedRow> addHostInfo
  ) {

    final List<StreamedRow> rows = new ArrayList<>();
    int begin = 0;

    for (int i = 0; i <= buff.length(); i++) {
      if ((i == buff.length() && (i - begin > 1))
          || (i < buff.length() && buff.getByte(i) == (byte) '\n')) {
        if (begin != i) { // Ignore random newlines - the server can send these
          final Buffer sliced = buff.slice(begin, i);
          final Buffer tidied = toJsonMsg(sliced, true);
          if (tidied.length() > 0) {
            final StreamedRow row = deserialize(tidied, StreamedRow.class);
            rows.add(addHostInfo.apply(row));
          }
        }

        begin = i + 1;
      }
    }
    return rows;
  }
@Test
  public void toRows_errorParsingNotAtEnd() {
    // When:
    final Exception e = assertThrows(
        KsqlRestClientException.class,
        () -> KsqlTargetUtil.toRows(Buffer.buffer(
            "[{\"header\":{\"queryId\":\"query_id_10\",\"schema\":\"`col1` STRING\"}},\n"
                + "{\"row\":{\"columns\"\n"
                + "{\"row\":{\"columns\":[\"Row2\"]}},\n"), Functions.identity())
    );

    // Then:
    assertThat(e.getMessage(), is(("Failed to deserialise object")));
  }