@Override
  public Type classify(final Throwable e) {
    Type type = Type.UNKNOWN;

    if (e instanceof KsqlSerializationException
        || (e instanceof StreamsException
            && (ExceptionUtils.indexOfThrowable(e, KsqlSerializationException.class) != -1))) {
      if (!hasInternalTopicPrefix(e)) {
        type = Type.USER;
        LOG.info(
            "Classified error as USER error based on schema mismatch. Query ID: {} Exception: {}",
            queryId,
            e);
      }
    }

    return type;
  }
@Test
  public void shouldClassifyWrappedKsqlSerializationExceptionWithChangelogTopicAsUnknownError() {
    // Given:
    final String topic = "_confluent-ksql-default_query_CTAS_USERS_0-Aggregate-Aggregate-Materialize-changelog";
    final Exception e = new StreamsException(
        new KsqlSerializationException(
            topic,
            "Error serializing message to topic: " + topic,
            new DataException("Struct schemas do not match.")));

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

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