public static void ensureCorrectArgs(
      final FunctionName functionName, final Object[] args, final Class<?>... argTypes
  ) {
    if (args == null) {
      throw new KsqlFunctionException("Null argument list for " + functionName.text() + ".");
    }

    if (args.length != argTypes.length) {
      throw new KsqlFunctionException("Incorrect arguments for " + functionName.text() + ".");
    }

    for (int i = 0; i < argTypes.length; i++) {
      if (args[i] == null) {
        continue;
      }

      if (!argTypes[i].isAssignableFrom(args[i].getClass())) {
        throw new KsqlFunctionException(
            String.format(
                "Incorrect arguments type for %s. "
                    + "Expected %s for arg number %d but found %s.",
                functionName.text(),
                argTypes[i].getCanonicalName(),
                i,
                args[i].getClass().getCanonicalName()
            ));
      }
    }
  }
@Test
  public void shouldHandleSubTypes() {
    final Object[] args = new Object[]{1.345, 55};
    UdfUtil.ensureCorrectArgs(FUNCTION_NAME, args, Number.class, Number.class);
  }