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 shouldParseError() {
    // When:
    final StreamedRow row = KsqlTargetUtil.toRowFromDelimited(Buffer.buffer(
        "{\"@type\":\"generic_error\",\"error_code\": 500,\"message\":\"Really bad problem\"}"));

    // Then:
    assertThat(row.getErrorMessage().isPresent(), is(true));
    assertThat(row.getErrorMessage().get().getErrorCode(), is(500));
    assertThat(row.getErrorMessage().get().getMessage(), is("Really bad problem"));
  }