@Override
  public synchronized void addAggregateFunctionFactory(
      final AggregateFunctionFactory aggregateFunctionFactory) {
    final String functionName = aggregateFunctionFactory.getName().toUpperCase();
    validateFunctionName(functionName);

    if (udfs.containsKey(functionName)) {
      throw new KsqlException(
          "Aggregate function already registered as non-aggregate: " + functionName);
    }

    if (udtfs.containsKey(functionName)) {
      throw new KsqlException(
          "Aggregate function already registered as table function: " + functionName);
    }

    if (udafs.putIfAbsent(functionName, aggregateFunctionFactory) != null) {
      throw new KsqlException("Aggregate function already registered: " + functionName);
    }

  }
@Test
  public void shouldThrowOnInvalidUdafFunctionName() {
    // Given:
    when(udafFactory.getName()).thenReturn("i am invalid");

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> functionRegistry.addAggregateFunctionFactory(udafFactory)
    );

    // Then:
    assertThat(e.getMessage(), containsString("is not a valid function name"));
  }