@Test
    public void testEmbeddingFunction() throws Exception {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(10000);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // step 1. Register custom function of Embedding
        tableEnv.createTemporaryFunction(DEFAULT_EMBEDDING_FUNCTION_NAME, EmbeddingFunction.class);

        List<String> udfNames = Arrays.asList(tableEnv.listUserDefinedFunctions());
        Assert.assertTrue(udfNames.contains(DEFAULT_EMBEDDING_FUNCTION_NAME.toLowerCase(Locale.ROOT)));

        // step 2. Generate test data and convert to DataStream
        int numOfMessages = 100;
        List<String> sourceDateList = new ArrayList<>();
        String msgPrefix = "Data for embedding-";
        for (int i = 0; i < numOfMessages; i++) {
            sourceDateList.add(msgPrefix + i);
        }

        List<Row> data = new ArrayList<>();
        sourceDateList.forEach(s -> data.add(Row.of(s)));
        TypeInformation<?>[] types = {BasicTypeInfo.STRING_TYPE_INFO};
        String[] names = {"f1"};
        RowTypeInfo typeInfo = new RowTypeInfo(types, names);
        DataStream<Row> dataStream = env.fromCollection(data).returns(typeInfo);

        // step 3. start a web server to mock embedding service
        String embeddingResult = "{\"result\": \"Result data for embedding\"}";
        HttpServer httpServer = HttpServer.create(new InetSocketAddress(8899), 0); // or use InetSocketAddress(0) for
                                                                                   // ephemeral port
        httpServer.createContext("/get_embedding", exchange -> {
            byte[] response = embeddingResult.getBytes(StandardCharsets.UTF_8);
            exchange.sendResponseHeaders(200, response.length);
            exchange.getResponseBody().write(response);
            exchange.close();
        });
        httpServer.start();

        // step 4. Convert from DataStream to Table and execute the Embedding function
        Table tempView = tableEnv.fromDataStream(dataStream).as("f1");
        tableEnv.createTemporaryView("temp_view", tempView);
        Table outputTable = tableEnv.sqlQuery(
                "SELECT " +
                        "f1," +
                        "EMBEDDING('http://localhost:8899/get_embedding', f1, 'BAAI/bge-large-en') " +
                        "from temp_view");

        // step 5. Get function execution result and parse it
        DataStream<Row> resultSet = tableEnv.toAppendStream(outputTable, Row.class);
        List<String> resultF0 = new ArrayList<>();
        List<String> resultF1 = new ArrayList<>();
        for (CloseableIterator<Row> it = resultSet.executeAndCollect(); it.hasNext();) {
            Row row = it.next();
            if (row != null) {
                resultF0.add(row.getField(0).toString());
                resultF1.add(row.getField(1).toString());
            }
        }
        Assert.assertEquals(resultF0.size(), numOfMessages);
        Assert.assertEquals(resultF1.size(), numOfMessages);
        Assert.assertEquals(resultF0, sourceDateList);
        for (String res : resultF1) {
            Assert.assertEquals(res, embeddingResult);
        }

        httpServer.stop(0);
    }
Test for embedding function

@throws Exception The exception may throw when test Embedding function
inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/function/EmbeddingFunctionTest.java