@VisibleForTesting
  static <T> Udaf<T, Struct, T> earliestT(
      final boolean ignoreNulls
  ) {
    return new Udaf<T, Struct, T>() {
      Schema structSchema;
      SqlType aggregateType;
      SqlType returnType;

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

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

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

      @Override
      public Struct initialize() {
        return null;
      }

      @Override
      public Struct aggregate(final T current, final Struct aggregate) {
        if (aggregate != null) {
          return aggregate;
        }

        if (current == null && ignoreNulls) {
          return null;
        }

        return createStruct(structSchema, current);
      }

      @Override
      public Struct merge(final Struct aggOne, final Struct aggTwo) {
        if (aggOne == null) {
          return aggTwo;
        }

        if (aggTwo == null) {
          return aggOne;
        }

        // When merging we need some way of evaluating the "earliest" one.
        // We do this by keeping track of the sequence of when it was originally processed
        if (INTERMEDIATE_STRUCT_COMPARATOR.compare(aggOne, aggTwo) < 0) {
          return aggOne;
        } else {
          return aggTwo;
        }
      }

      @Override
      @SuppressWarnings("unchecked")
      public T map(final Struct agg) {
        if (agg == null) {
          return null;
        }

        return (T) agg.get(VAL_FIELD);
      }
    };
  }
@Test
  public void shouldInitialize() {
    // Given:
    final Udaf<Integer, Struct, Integer> udaf = EarliestByOffset
        .earliestT(true);

    // When:
    final Struct init = udaf.initialize();

    // Then:
    assertThat(init, is(nullValue()));
  }