@VisibleForTesting
  static <T> Udaf<T, List<Struct>, List<T>> earliestTN(
      final int earliestN,
      final boolean ignoreNulls
  ) {
    if (earliestN <= 0) {
      throw new KsqlFunctionException("earliestN must be 1 or greater");
    }

    return new Udaf<T, List<Struct>, List<T>>() {
      Schema structSchema;
      SqlType aggregateType;
      SqlType returnType;

      @Override
      public void initializeTypeArguments(final List<SqlArgument> argTypeList) {
        final SqlType inputType = argTypeList.get(0).getSqlTypeOrThrow();
        final Schema connectType =
            SchemaConverters.sqlToConnectConverter().toConnectSchema(inputType);
        structSchema = KudafByOffsetUtils.buildSchema(connectType);
        aggregateType =
            SqlArray.of(SchemaConverters.connectToSqlConverter().toSqlType(structSchema));
        returnType = SqlArray.of(inputType);
      }

      @Override
      public Optional<SqlType> getAggregateSqlType() {
        return Optional.of(aggregateType);
      }

      @Override
      public Optional<SqlType> getReturnSqlType() {
        return Optional.of(returnType);
      }

      @Override
      public List<Struct> initialize() {
        return new ArrayList<>(earliestN);
      }

      @Override
      public List<Struct> aggregate(final T current, final List<Struct> aggregate) {
        if (current == null && ignoreNulls) {
          return aggregate;
        }

        if (aggregate.size() < earliestN) {
          aggregate.add(createStruct(structSchema, current));
        }
        return aggregate;
      }

      @Override
      public List<Struct> merge(final List<Struct> aggOne, final List<Struct> aggTwo) {
        final List<Struct> merged = new ArrayList<>(aggOne.size() + aggTwo.size());
        merged.addAll(aggOne);
        merged.addAll(aggTwo);
        merged.sort(INTERMEDIATE_STRUCT_COMPARATOR);
        return merged.subList(0, Math.min(earliestN, merged.size()));
      }

      @Override
      @SuppressWarnings("unchecked")
      public List<T> map(final List<Struct> agg) {
        return (List<T>) agg.stream().map(s -> s.get(VAL_FIELD)).collect(Collectors.toList());
      }
    };
  }
@Test
  public void shouldThrowExceptionForInvalidN() {
    try {
      EarliestByOffset.earliestTN( -1, true);
    } catch (KsqlException e) {
      assertThat(e.getMessage(), is("earliestN must be 1 or greater"));
    }
  }