@Override
  public Type classify(final Throwable e) {
    Type type = Type.UNKNOWN;
    if (e instanceof KsqlFunctionException
        || (e instanceof StreamsException
        && ExceptionUtils.getRootCause(e) instanceof KsqlFunctionException)) {
      type = Type.USER;
    }

    if (type == Type.USER) {
      LOG.info(
          "Classified error as USER error based on invalid user input. Query ID: {} Exception: {}",
          queryId,
          e);
    }

    return type;
  }
@Test
  public void shouldClassifyStreamsExceptionAsUnknownError() {
    // Given:
    final Exception e = new StreamsException("foo");

    // When:
    final Type type = new KsqlFunctionClassifier("").classify(e);

    // Then:
    assertThat(type, is(Type.UNKNOWN));
  }