@Override
  public synchronized UdfFactory ensureFunctionFactory(final UdfFactory factory) {
    validateFunctionName(factory.getName());

    final String functionName = factory.getName().toUpperCase();
    if (udafs.containsKey(functionName)) {
      throw new KsqlException("UdfFactory already registered as aggregate: " + functionName);
    }
    if (udtfs.containsKey(functionName)) {
      throw new KsqlException("UdfFactory already registered as table function: " + functionName);
    }

    final UdfFactory existing = udfs.putIfAbsent(functionName, factory);
    if (existing != null && !existing.matches(factory)) {
      throw new KsqlException("UdfFactory not compatible with existing factory."
          + " function: " + functionName
          + " existing: " + existing
          + ", factory: " + factory);
    }

    return existing == null ? factory : existing;
  }
@Test
  public void shouldThrowOnEnsureUdfFactoryOnDifferentX() {
    // Given:
    functionRegistry.ensureFunctionFactory(udfFactory);

    when(udfFactory.matches(udfFactory1)).thenReturn(false);

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

    // Then:
    assertThat(e.getMessage(), containsString("UdfFactory not compatible with existing factory"));
  }