public UserDefinedFunctionDescriptor(String name, String classpath) {
        this.name = name;
        this.classpath = classpath;
        this.className = classpath.substring(classpath.lastIndexOf('.') + 1);
        try {
            Class<?> clazz = Class.forName(classpath);
            isCdcPipelineUdf = isCdcPipelineUdf(clazz);
            if (isCdcPipelineUdf) {
                // We use reflection to invoke UDF methods since we may add more methods
                // into UserDefinedFunction interface, thus the provided UDF classes
                // might not be compatible with the interface definition in CDC common.
                returnTypeHint =
                        (DataType) clazz.getMethod("getReturnType").invoke(clazz.newInstance());
            } else {
                returnTypeHint = null;
            }
        } catch (ClassNotFoundException
                | InvocationTargetException
                | IllegalAccessException
                | NoSuchMethodException
                | InstantiationException e) {
            throw new IllegalArgumentException(
                    "Failed to instantiate UDF " + name + "@" + classpath, e);
        }
    }
@Test
    void testUserDefinedFunctionDescriptor() {

        assertThat(new UserDefinedFunctionDescriptor("cdc_udf", CdcUdf.class.getName()))
                .extracting("name", "className", "classpath", "returnTypeHint", "isCdcPipelineUdf")
                .containsExactly(
                        "cdc_udf",
                        "UserDefinedFunctionDescriptorTest$CdcUdf",
                        "org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptorTest$CdcUdf",
                        null,
                        true);

        assertThat(
                        new UserDefinedFunctionDescriptor(
                                "cdc_udf_with_type_hint", CdcUdfWithTypeHint.class.getName()))
                .extracting("name", "className", "classpath", "returnTypeHint", "isCdcPipelineUdf")
                .containsExactly(
                        "cdc_udf_with_type_hint",
                        "UserDefinedFunctionDescriptorTest$CdcUdfWithTypeHint",
                        "org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptorTest$CdcUdfWithTypeHint",
                        DataTypes.TIMESTAMP_LTZ(9),
                        true);

        assertThat(new UserDefinedFunctionDescriptor("flink_udf", FlinkUdf.class.getName()))
                .extracting("name", "className", "classpath", "returnTypeHint", "isCdcPipelineUdf")
                .containsExactly(
                        "flink_udf",
                        "UserDefinedFunctionDescriptorTest$FlinkUdf",
                        "org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptorTest$FlinkUdf",
                        null,
                        false);

        assertThatThrownBy(
                        () -> new UserDefinedFunctionDescriptor("not_udf", NotUDF.class.getName()))
                .isExactlyInstanceOf(IllegalArgumentException.class)
                .hasMessage(
                        "Failed to detect UDF class class org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptorTest$NotUDF "
                                + "since it never implements interface org.apache.flink.cdc.common.udf.UserDefinedFunction or "
                                + "extends Flink class org.apache.flink.table.functions.ScalarFunction.");

        assertThatThrownBy(
                        () ->
                                new UserDefinedFunctionDescriptor(
                                        "not_even_exist", "not.a.valid.class.path"))
                .isExactlyInstanceOf(IllegalArgumentException.class)
                .hasMessage("Failed to instantiate UDF not_even_exist@not.a.valid.class.path");
    }