[FLINK-36525] address comments.

Author: lvyanquan

docs/content/docs/core-concept/transform.md

@@ -415,7 +415,7 @@ The following built-in models are provided:
 | parameter          | type   | optional/required | meaning                                                                                                                              |
 |--------------------|--------|-------------------|--------------------------------------------------------------------------------------------------------------------------------------|
 | openai.model       | STRING | required          | Name of model to be called, for example: "gpt-4o-mini", Available options are "gpt-4o-mini", "gpt-4o", "gpt-4-32k", "gpt-3.5-turbo". |
-| openai.host        | STRING | required          | Host of the Model server to be connected, for example: "http://langchain4j.dev/demo/openai/v1".                                      |
+| openai.host        | STRING | required          | Host of the Model server to be connected, for example: `http://langchain4j.dev/demo/openai/v1`.                                      |
 | openai.apikey      | STRING | required          | Api Key for verification of the Model server, for example, "demo".                                                                   |
 | openai.chat.prompt | STRING | optional          | Prompt for chatting with OpenAI, for example: "Please summary this ".                                                                |
 
@@ -424,7 +424,7 @@ The following built-in models are provided:
 | parameter     | type   | optional/required | meaning                                                                                                                                                                |
 |---------------|--------|-------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | openai.model  | STRING | required          | Name of model to be called, for example: "text-embedding-3-small", Available options are "text-embedding-3-small", "text-embedding-3-large", "text-embedding-ada-002". |
-| openai.host   | STRING | required          | Host of the Model server to be connected, for example: "http://langchain4j.dev/demo/openai/v1".                                                                        |
+| openai.host   | STRING | required          | Host of the Model server to be connected, for example: `http://langchain4j.dev/demo/openai/v1`.                                                                        |
 | openai.apikey | STRING | required          | Api Key for verification of the Model server, for example, "demo".                                                                                                     |
 
 

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java

@@ -38,6 +38,10 @@
  */
 public class TransformTranslator {
 
+    /** Package of built-in model. */
+    public static final String PREFIX_CLASSPATH_BUILT_IN_MODEL =
+            "org.apache.flink.cdc.runtime.model.";
+
     public DataStream<Event> translatePreTransform(
             DataStream<Event> input,
             List<TransformDef> transforms,
@@ -100,7 +104,10 @@ public DataStream<Event> translatePostTransform(
     }
 
     private Tuple3<String, String, Map<String, String>> modelToUDFTuple(ModelDef model) {
-        return Tuple3.of(model.getModelName(), model.getClassName(), model.getParameters());
+        return Tuple3.of(
+                model.getModelName(),
+                PREFIX_CLASSPATH_BUILT_IN_MODEL + model.getClassName(),
+                model.getParameters());
     }
 
     private Tuple3<String, String, Map<String, String>> udfDefToUDFTuple(UdfDef udf) {

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java

@@ -829,7 +829,7 @@ void testComplicatedFlinkUdf(ValuesDataSink.SinkApi sinkApi, String language) th
 
     @ParameterizedTest
     @MethodSource("testParams")
-    void testTransformWithModel(ValuesDataSink.SinkApi sinkApi, String language) throws Exception {
+    void testTransformWithModel(ValuesDataSink.SinkApi sinkApi) throws Exception {
         FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
 
         // Setup value source

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/UserDefinedFunctionDescriptor.java

@@ -40,30 +40,20 @@ public class UserDefinedFunctionDescriptor implements Serializable {
     private final String className;
     private final DataType returnTypeHint;
     private final boolean isCdcPipelineUdf;
-
     private final Map<String, String> parameters;
 
-    /** Package of built-in model. */
-    public static final String PREFIX_CLASSPATH_BUILT_IN_MODEL =
-            "org.apache.flink.cdc.runtime.model.";
-
     public UserDefinedFunctionDescriptor(String name, String classpath) {
         this(name, classpath, new HashMap<>());
     }
 
     public UserDefinedFunctionDescriptor(
             String name, String classpath, Map<String, String> parameters) {
-        if (classpath.contains(".")) {
-            this.className = classpath.substring(classpath.lastIndexOf('.') + 1);
-            this.classpath = classpath;
-        } else {
-            this.className = classpath;
-            this.classpath = PREFIX_CLASSPATH_BUILT_IN_MODEL + classpath;
-        }
         this.name = name;
         this.parameters = parameters;
+        this.classpath = classpath;
+        this.className = classpath.substring(classpath.lastIndexOf('.') + 1);
         try {
-            Class<?> clazz = Class.forName(this.classpath);
+            Class<?> clazz = Class.forName(classpath);
             isCdcPipelineUdf = isCdcPipelineUdf(clazz);
             if (isCdcPipelineUdf) {
                 // We use reflection to invoke UDF methods since we may add more methods