Repository: risingwavelabs/risingwave (language: python)

Issue:
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/java/connector-node/risingwave-connector-service/pom.xml b/java/connector-node/risingwave-connector-service/pom.xml
index c2ce04c82691d..eac42aedc1956 100644
--- a/java/connector-node/risingwave-connector-service/pom.xml
+++ b/java/connector-node/risingwave-connector-service/pom.xml
@@ -111,13 +111,5 @@
             <artifactId>risingwave-sink-cassandra</artifactId>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>risingwave-sink-mock-flink-runtime</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>risingwave-sink-mock-flink-http-sink</artifactId>
-        </dependency>
     </dependencies>
 </project>
diff --git a/java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkUtils.java b/java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkUtils.java
index 4d274a2ef20b6..776f9c179771b 100644
--- a/java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkUtils.java
+++ b/java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkUtils.java
@@ -17,8 +17,6 @@
 import static io.grpc.Status.*;
 
 import com.risingwave.connector.api.sink.SinkFactory;
-import com.risingwave.mock.flink.http.HttpFlinkMockSinkFactory;
-import com.risingwave.mock.flink.runtime.FlinkDynamicAdapterFactory;
 import com.risingwave.proto.ConnectorServiceProto;
 import java.util.Optional;
 
@@ -44,8 +42,6 @@ public static SinkFactory getSinkFactory(String sinkName) {
                 return new EsSinkFactory();
             case "cassandra":
                 return new CassandraFactory();
-            case "http":
-                return new FlinkDynamicAdapterFactory(new HttpFlinkMockSinkFactory());
             default:
                 throw UNIMPLEMENTED
                         .withDescription("unknown sink type: " + sinkName)
diff --git a/java/connector-node/risingwave-sink-mock-flink/pom.xml b/java/connector-node/risingwave-sink-mock-flink/pom.xml
deleted file mode 100644
index 548fd9d78f442..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/pom.xml
+++ /dev/null
@@ -1,47 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>risingwave-java-root</artifactId>
-        <groupId>com.risingwave</groupId>
-        <version>0.1.0-SNAPSHOT</version>
-        <relativePath>../../pom.xml</relativePath>
-    </parent>
-    <packaging>pom</packaging>
-    <modules>
-        <module>risingwave-sink-mock-flink-runtime</module>
-        <module>risingwave-sink-mock-flink-common</module>
-        <module>risingwave-sink-mock-flink-http-sink</module>
-    </modules>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>risingwave-sink-mock-flink</artifactId>
-    <version>0.1.0-SNAPSHOT</version>
-    <name>risingwave-sink-dynamic</name>
-
-    <properties>
-        <flink.version>1.18.0</flink.version>
-        <maven.javadoc.skip>true</maven.javadoc.skip>
-    </properties>
-
-    <dependencyManagement>
-        <dependencies>
-            <dependency>
-                <groupId>org.apache.flink</groupId>
-                <artifactId>flink-table-common</artifactId>
-                <version>${flink.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>org.apache.flink</groupId>
-                <artifactId>flink-json</artifactId>
-                <version>${flink.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>org.apache.flink</groupId>
-                <artifactId>flink-connector-base</artifactId>
-                <version>${flink.version}</version>
-            </dependency>
-        </dependencies>
-    </dependencyManagement>
-</project>
\ No newline at end of file
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/pom.xml b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/pom.xml
deleted file mode 100644
index d2d1c44d825ed..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/pom.xml
+++ /dev/null
@@ -1,31 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
-    <parent>
-        <groupId>com.risingwave</groupId>
-        <artifactId>risingwave-sink-mock-flink</artifactId>
-        <version>0.1.0-SNAPSHOT</version>
-    </parent>
-
-    <groupId>com.risingwave</groupId>
-    <artifactId>risingwave-sink-mock-flink-common</artifactId>
-
-    <properties>
-        <maven.compiler.source>11</maven.compiler.source>
-        <maven.compiler.target>11</maven.compiler.target>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    </properties>
-    <dependencies>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>connector-api</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-common</artifactId>
-        </dependency>
-    </dependencies>
-
-</project>
\ No newline at end of file
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkDynamicAdapterConfig.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkDynamicAdapterConfig.java
deleted file mode 100644
index 82d6f66bb2991..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkDynamicAdapterConfig.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.common;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.risingwave.connector.api.sink.CommonSinkConfig;
-import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.table.catalog.ObjectPath;
-
-/*
- * The user-input configuration items will be passed downstream to create the corresponding factory.
- */
-public class FlinkDynamicAdapterConfig extends CommonSinkConfig {
-    Map<String, String> option;
-
-    @JsonCreator
-    public FlinkDynamicAdapterConfig(Map<String, String> tableProperties) {
-        super(
-                tableProperties.get("connector"),
-                Boolean.valueOf(tableProperties.get("force_append_only")),
-                tableProperties.get("primary_key"));
-        this.option = tableProperties;
-    }
-
-    public ObjectPath getTablePath() {
-        throw new RuntimeException("Cannot get table with connector type " + getConnector());
-    }
-
-    /**
-     * Our option contains both kinds of parameters, and it is the responsibility of the correct
-     * parameter to be chosen here
-     *
-     * @param needOptionSet: Need option names
-     */
-    public void processOption(Set<ConfigOption<?>> needOptionSet) {
-        Set<String> needOptionStringSet =
-                needOptionSet.stream().map(c -> c.key()).collect(Collectors.toSet());
-        option =
-                option.entrySet().stream()
-                        .filter(entry -> needOptionStringSet.contains(entry.getKey()))
-                        .collect(Collectors.toMap(a -> a.getKey(), a -> a.getValue()));
-    }
-
-    public Map<String, String> getOption() {
-        return option;
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkDynamicUtil.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkDynamicUtil.java
deleted file mode 100644
index 4a58c3160b075..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkDynamicUtil.java
+++ /dev/null
@@ -1,200 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.common;
-
-import com.risingwave.connector.api.ColumnDesc;
-import com.risingwave.connector.api.TableSchema;
-import com.risingwave.proto.Data;
-import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.api.Schema;
-import org.apache.flink.table.catalog.Catalog;
-import org.apache.flink.table.catalog.Column;
-import org.apache.flink.table.catalog.exceptions.TableNotExistException;
-import org.apache.flink.table.factories.CatalogFactory;
-import org.apache.flink.table.factories.FactoryUtil;
-import org.apache.flink.table.types.DataType;
-
-public class FlinkDynamicUtil {
-    private static final String TIMESTAMP_LTZ_FORMAT = "TIMESTAMP\\(\\d+\\) WITH LOCAL TIME ZONE";
-
-    public static DataType getCorrespondingFlinkType(Data.DataType dataType) {
-        switch (dataType.getTypeName()) {
-            case INT16:
-                return DataTypes.SMALLINT();
-            case INT32:
-                return DataTypes.INT();
-            case INT64:
-                return DataTypes.BIGINT();
-            case FLOAT:
-                return DataTypes.FLOAT();
-            case DOUBLE:
-                return DataTypes.DOUBLE();
-            case BOOLEAN:
-                return DataTypes.BOOLEAN();
-            case VARCHAR:
-                return DataTypes.STRING();
-            case DECIMAL:
-                // Because there is no 'scale' and 'precision' in 'DECIMAL' within RW,
-                // this is just a simulation here, which may be incompatible with some interfaces.
-                return DataTypes.DECIMAL(9, 1);
-            case TIMESTAMP:
-                // Like DECIMAL
-                return DataTypes.TIMESTAMP();
-            case TIMESTAMPTZ:
-                // Like DECIMAL
-                return DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE();
-            case DATE:
-                return DataTypes.DATE();
-            case TIME:
-                return DataTypes.TIME();
-            case BYTEA:
-                return DataTypes.BYTES();
-            case LIST:
-            case STRUCT:
-            case INTERVAL:
-                throw Status.UNIMPLEMENTED
-                        .withDescription(String.format("not support %s now", dataType))
-                        .asRuntimeException();
-            default:
-                throw Status.INVALID_ARGUMENT
-                        .withDescription("unspecified type" + dataType)
-                        .asRuntimeException();
-        }
-    }
-
-    /** Converting RW's schema to Flink's schema */
-    public static List<Column> getFlinkColumnsFromSchema(TableSchema tableSchema) {
-        List<Column> columns = new ArrayList<>();
-        for (ColumnDesc columnDesc : tableSchema.getColumnDescs()) {
-            columns.add(
-                    Column.physical(
-                            columnDesc.getName(),
-                            getCorrespondingFlinkType(columnDesc.getDataType())));
-        }
-        return columns;
-    }
-
-    public static void validateSchemaWithCatalog(
-            TableSchema tableSchema,
-            FlinkDynamicAdapterConfig config,
-            CatalogFactory catalogFactory)
-            throws StatusRuntimeException {
-        List<Column> flinkColumns = getColumns(tableSchema, config, catalogFactory);
-        List<ColumnDesc> rwColumnDescs = tableSchema.getColumnDescs();
-        if (rwColumnDescs.size() != flinkColumns.size()) {
-            throw Status.FAILED_PRECONDITION
-                    .withDescription("Don't match in the number of columns in the table")
-                    .asRuntimeException();
-        }
-
-        Map<String, DataType> flinkColumnMap =
-                flinkColumns.stream()
-                        .collect(Collectors.toMap(e -> e.getName(), e -> e.getDataType()));
-        for (ColumnDesc columnDesc : rwColumnDescs) {
-            if (!flinkColumnMap.containsKey(columnDesc.getName())) {
-                throw Status.FAILED_PRECONDITION
-                        .withDescription(
-                                String.format(
-                                        "Name mismatch. Column `%s` on RisingWave side is not found on Flink side.",
-                                        columnDesc.getName()))
-                        .asRuntimeException();
-            }
-            if (!checkType(columnDesc.getDataType(), flinkColumnMap.get(columnDesc.getName()))) {
-                throw Status.FAILED_PRECONDITION
-                        .withDescription(
-                                String.format(
-                                        "Data type mismatch for column `%s`. Flink side: `%s`, RisingWave side: `%s`.",
-                                        columnDesc.getName(),
-                                        flinkColumnMap.get(columnDesc.getName()),
-                                        columnDesc.getDataType().getTypeName()))
-                        .asRuntimeException();
-            }
-        }
-    }
-
-    private static List<Column> getColumns(
-            TableSchema tableSchema,
-            FlinkDynamicAdapterConfig config,
-            CatalogFactory catalogFactory) {
-        ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
-        Configuration configuration = new Configuration();
-        Set<ConfigOption<?>> configOptions = catalogFactory.requiredOptions();
-        configOptions.addAll(catalogFactory.optionalOptions());
-        config.processOption(configOptions);
-
-        Schema.Builder schemaBuilder = Schema.newBuilder();
-        tableSchema
-                .getColumnDescs()
-                .forEach(
-                        (columnDesc) ->
-                                schemaBuilder.column(
-                                        columnDesc.getName(),
-                                        getCorrespondingFlinkType(columnDesc.getDataType())));
-        // Start with the default value, and add as needed later
-        FactoryUtil.DefaultCatalogContext defaultCatalogContext =
-                new FactoryUtil.DefaultCatalogContext(
-                        "catalog", config.getOption(), configuration, contextClassLoader);
-        Catalog catalog = catalogFactory.createCatalog(defaultCatalogContext);
-
-        List<Column> columns = new ArrayList<>();
-        try {
-            catalog.open();
-            List<Schema.UnresolvedColumn> flinkColumns =
-                    catalog.getTable(config.getTablePath()).getUnresolvedSchema().getColumns();
-            catalog.close();
-            for (int i = 0; i < flinkColumns.size(); i++) {
-                Schema.UnresolvedColumn unresolvedColumn = flinkColumns.get(i);
-                if (unresolvedColumn instanceof Schema.UnresolvedPhysicalColumn) {
-                    Schema.UnresolvedPhysicalColumn c1 =
-                            (Schema.UnresolvedPhysicalColumn) unresolvedColumn;
-                    columns.add(Column.physical(c1.getName(), (DataType) (c1.getDataType())));
-                } else {
-                    throw new RuntimeException("Only support physical column");
-                }
-            }
-            return columns;
-        } catch (TableNotExistException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    private static boolean checkType(Data.DataType dataType, DataType flinkType) {
-        switch (dataType.getTypeName()) {
-            case DECIMAL:
-                return flinkType.toString().contains("DECIMAL");
-            case TIMESTAMP:
-                return flinkType.toString().contains("TIMESTAMP");
-            case TIMESTAMPTZ:
-                Pattern regex = Pattern.compile(TIMESTAMP_LTZ_FORMAT);
-                Matcher matcher = regex.matcher(flinkType.toString());
-                return matcher.matches();
-            default:
-                return getCorrespondingFlinkType(dataType).equals(flinkType);
-        }
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkMockSinkFactory.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkMockSinkFactory.java
deleted file mode 100644
index 0d0e716d56da0..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkMockSinkFactory.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.common;
-
-import com.risingwave.connector.api.TableSchema;
-import io.grpc.StatusRuntimeException;
-import org.apache.flink.table.factories.DynamicTableSinkFactory;
-
-/**
- * Different sinks need to implement this method to provide the required `validate` and
- * `DynamicTableSinkFactory` for flink sinks
- */
-public interface FlinkMockSinkFactory {
-    /**
-     * It is responsible for validating our schema, the default use of the flink catalog interface.
-     * But some sinks do not implement the catalog, we need to implement their own checksums
-     */
-    public void validate(TableSchema tableSchema, FlinkDynamicAdapterConfig config)
-            throws StatusRuntimeException;
-
-    public DynamicTableSinkFactory getDynamicTableSinkFactory();
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/pom.xml b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/pom.xml
deleted file mode 100644
index 36741240134ed..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/pom.xml
+++ /dev/null
@@ -1,60 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
-    <parent>
-        <groupId>com.risingwave</groupId>
-        <artifactId>risingwave-sink-mock-flink</artifactId>
-        <version>0.1.0-SNAPSHOT</version>
-    </parent>
-
-    <groupId>com.risingwave</groupId>
-    <artifactId>risingwave-sink-mock-flink-http-sink</artifactId>
-
-    <properties>
-        <maven.compiler.source>11</maven.compiler.source>
-        <maven.compiler.target>11</maven.compiler.target>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    </properties>
-    <dependencies>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>risingwave-sink-mock-flink-common</artifactId>
-        </dependency>
-        <!-- https://mvnrepository.com/artifact/com.getindata/flink-http-connector -->
-        <dependency>
-            <groupId>com.getindata</groupId>
-            <artifactId>flink-http-connector</artifactId>
-            <version>0.16.0</version>
-        </dependency>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>risingwave-sink-mock-flink-runtime</artifactId>
-            <scope>provided</scope>
-        </dependency>
-
-        <!-- test dependencies -->
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-api</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-slf4j2-impl</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-core</artifactId>
-            <scope>test</scope>
-        </dependency>
-    </dependencies>
-
-</project>
\ No newline at end of file
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/src/main/java/com/risingwave/mock/flink/http/HttpFlinkMockSinkFactory.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/src/main/java/com/risingwave/mock/flink/http/HttpFlinkMockSinkFactory.java
deleted file mode 100644
index d316eeae74bed..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/src/main/java/com/risingwave/mock/flink/http/HttpFlinkMockSinkFactory.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.http;
-
-import com.getindata.connectors.http.internal.table.sink.HttpDynamicTableSinkFactory;
-import com.risingwave.connector.api.TableSchema;
-import com.risingwave.mock.flink.common.FlinkDynamicAdapterConfig;
-import com.risingwave.mock.flink.common.FlinkMockSinkFactory;
-import io.grpc.StatusRuntimeException;
-import org.apache.flink.table.factories.DynamicTableSinkFactory;
-
-/**
- * The `FlinkMockSinkFactory` implementation of the http sink is responsible for creating the http
- * counterpart of the `DynamicTableSinkFactory`. And `validate` don't need to do anything.
- *
- * <p>This feature is depended on https://github.com/getindata/flink-http-connector
- */
-public class HttpFlinkMockSinkFactory implements FlinkMockSinkFactory {
-    @Override
-    public void validate(TableSchema tableSchema, FlinkDynamicAdapterConfig config)
-            throws StatusRuntimeException {}
-
-    @Override
-    public DynamicTableSinkFactory getDynamicTableSinkFactory() {
-        return new HttpDynamicTableSinkFactory();
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/pom.xml b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/pom.xml
deleted file mode 100644
index 00f93908a6f24..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/pom.xml
+++ /dev/null
@@ -1,84 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
-    <parent>
-        <groupId>com.risingwave</groupId>
-        <artifactId>risingwave-sink-mock-flink</artifactId>
-        <version>0.1.0-SNAPSHOT</version>
-    </parent>
-
-    <groupId>com.risingwave</groupId>
-    <artifactId>risingwave-sink-mock-flink-runtime</artifactId>
-
-    <properties>
-        <maven.compiler.source>11</maven.compiler.source>
-        <maven.compiler.target>11</maven.compiler.target>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    </properties>
-
-    <dependencies>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>proto</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>connector-api</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.commons</groupId>
-            <artifactId>commons-text</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-databind</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-core</artifactId>
-        </dependency>
-
-        <!-- flink drivers -->
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-common</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-json</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>risingwave-sink-mock-flink-common</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-base</artifactId>
-        </dependency>
-
-        <!-- test dependencies -->
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-api</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-slf4j2-impl</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-core</artifactId>
-            <scope>test</scope>
-        </dependency>
-    </dependencies>
-
-</project>
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/FlinkDynamicAdapterFactory.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/FlinkDynamicAdapterFactory.java
deleted file mode 100644
index f9ebebef82743..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/FlinkDynamicAdapterFactory.java
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.risingwave.connector.api.TableSchema;
-import com.risingwave.connector.api.sink.SinkFactory;
-import com.risingwave.connector.api.sink.SinkWriter;
-import com.risingwave.mock.flink.common.FlinkDynamicAdapterConfig;
-import com.risingwave.mock.flink.common.FlinkDynamicUtil;
-import com.risingwave.mock.flink.common.FlinkMockSinkFactory;
-import com.risingwave.mock.flink.runtime.context.DynamicTableSinkContextImpl;
-import com.risingwave.mock.flink.runtime.context.SinkWriterContextV2;
-import com.risingwave.mock.flink.runtime.sinkwriter.AsyncSinkWriterImpl;
-import com.risingwave.mock.flink.runtime.sinkwriter.SinkWriterImpl;
-import com.risingwave.mock.flink.runtime.sinkwriter.SinkWriterV2Impl;
-import com.risingwave.proto.Catalog;
-import java.io.IOException;
-import java.util.*;
-import org.apache.flink.api.connector.sink.Sink;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.base.sink.writer.AsyncSinkWriter;
-import org.apache.flink.table.api.Schema;
-import org.apache.flink.table.api.TableException;
-import org.apache.flink.table.catalog.*;
-import org.apache.flink.table.connector.sink.*;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.factories.DynamicTableSinkFactory;
-import org.apache.flink.table.factories.FactoryUtil;
-
-public class FlinkDynamicAdapterFactory implements SinkFactory {
-
-    FlinkMockSinkFactory flinkMockSinkFactory;
-
-    public FlinkDynamicAdapterFactory(FlinkMockSinkFactory flinkMockSinkFactory) {
-        this.flinkMockSinkFactory = flinkMockSinkFactory;
-    }
-
-    // Create corresponding Sink according to different configurations.
-    public DynamicTableSink.SinkRuntimeProvider buildDynamicTableSinkProvider(
-            TableSchema tableSchema, FlinkDynamicAdapterConfig config) {
-        List<Column> flinkColumns = FlinkDynamicUtil.getFlinkColumnsFromSchema(tableSchema);
-
-        // Start with the default value, and add as needed later
-        ObjectIdentifier objectIdentifier =
-                ObjectIdentifier.of("catalog_name", "database_name", "obj_name");
-        ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
-        DynamicTableSinkFactory dynamicTableSinkFactory =
-                flinkMockSinkFactory.getDynamicTableSinkFactory();
-        Set<ConfigOption<?>> configOptions = new HashSet<>();
-        configOptions.addAll(
-                dynamicTableSinkFactory.requiredOptions() == null
-                        ? Collections.emptyList()
-                        : dynamicTableSinkFactory.requiredOptions());
-        configOptions.addAll(
-                dynamicTableSinkFactory.optionalOptions() == null
-                        ? Collections.emptyList()
-                        : dynamicTableSinkFactory.optionalOptions());
-        config.processOption(configOptions);
-
-        Schema.Builder schemaBuilder = Schema.newBuilder();
-        flinkColumns.forEach(
-                (column) -> schemaBuilder.column(column.getName(), column.getDataType()));
-        Schema shcema = schemaBuilder.primaryKey(tableSchema.getPrimaryKeys()).build();
-
-        ResolvedCatalogTable resolvedCatalogTable =
-                new ResolvedCatalogTable(
-                        CatalogTable.of(
-                                shcema, null, tableSchema.getPrimaryKeys(), config.getOption()),
-                        ResolvedSchema.of(flinkColumns));
-
-        FactoryUtil.DefaultDynamicTableContext defaultDynamicTableContext =
-                new FactoryUtil.DefaultDynamicTableContext(
-                        objectIdentifier,
-                        resolvedCatalogTable,
-                        new HashMap<String, String>(),
-                        new Configuration(),
-                        contextClassLoader,
-                        false);
-        DynamicTableSink.Context context = new DynamicTableSinkContextImpl();
-        // Start with the default value, and add as needed later
-        return dynamicTableSinkFactory
-                .createDynamicTableSink(defaultDynamicTableContext)
-                .getSinkRuntimeProvider(context);
-    }
-
-    /**
-     * According to different implementations, create different sink writer, and package it as a
-     * RW-recognizable method(SinkWriterV1), after which the RW can call the method to complete
-     * write directly.
-     */
-    @Override
-    public SinkWriter createWriter(TableSchema tableSchema, Map<String, String> tableProperties) {
-        ObjectMapper mapper = new ObjectMapper();
-        FlinkDynamicAdapterConfig flinkDynamicAdapterConfig =
-                mapper.convertValue(tableProperties, FlinkDynamicAdapterConfig.class);
-
-        DynamicTableSink.SinkRuntimeProvider sinkRuntimeProvider =
-                buildDynamicTableSinkProvider(tableSchema, flinkDynamicAdapterConfig);
-        if (sinkRuntimeProvider instanceof SinkProvider) {
-            Sink<RowData, ?, ?, ?> sink = ((SinkProvider) sinkRuntimeProvider).createSink();
-            return new SinkWriterImpl<>(tableSchema, sink);
-        } else if (sinkRuntimeProvider instanceof SinkV2Provider) {
-            org.apache.flink.api.connector.sink2.Sink<RowData> sink =
-                    ((SinkV2Provider) sinkRuntimeProvider).createSink();
-
-            try {
-                if (sink.createWriter(new SinkWriterContextV2()) instanceof AsyncSinkWriter) {
-                    return new AsyncSinkWriterImpl(tableSchema, sink);
-                }
-            } catch (IOException e) {
-                throw new RuntimeException(e);
-            }
-
-            return new SinkWriterV2Impl(tableSchema, sink);
-        } else {
-            throw new TableException("Unsupported sink runtime provider.");
-        }
-    }
-
-    @Override
-    public void validate(
-            TableSchema tableSchema,
-            Map<String, String> tableProperties,
-            Catalog.SinkType sinkType) {
-        ObjectMapper mapper = new ObjectMapper();
-        FlinkDynamicAdapterConfig flinkDynamicAdapterConfig =
-                mapper.convertValue(tableProperties, FlinkDynamicAdapterConfig.class);
-        flinkMockSinkFactory.validate(tableSchema, flinkDynamicAdapterConfig);
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/RowDataImpl.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/RowDataImpl.java
deleted file mode 100644
index 7129fdd1e1900..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/RowDataImpl.java
+++ /dev/null
@@ -1,168 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime;
-
-import com.risingwave.connector.api.sink.SinkRow;
-import io.grpc.Status;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.OffsetDateTime;
-import java.util.ArrayList;
-import java.util.Base64;
-import org.apache.flink.table.data.*;
-import org.apache.flink.types.RowKind;
-
-/*
- * Converts data types from RW to Flink's RowData (Format of flink writes to sink).
- */
-public class RowDataImpl implements RowData {
-
-    private SinkRow sinkRow;
-    private int arity;
-
-    public RowDataImpl(SinkRow sinkRow, int arity) {
-        this.sinkRow = sinkRow;
-        this.arity = arity;
-    }
-
-    @Override
-    public int getArity() {
-        return arity;
-    }
-
-    @Override
-    public RowKind getRowKind() {
-        switch (sinkRow.getOp()) {
-            case INSERT:
-                return RowKind.INSERT;
-            case UPDATE_INSERT:
-                return RowKind.UPDATE_AFTER;
-            case UPDATE_DELETE:
-                return RowKind.UPDATE_BEFORE;
-            case DELETE:
-                return RowKind.DELETE;
-            default:
-                throw Status.INTERNAL
-                        .withDescription("Unknown operation: " + sinkRow.getOp())
-                        .asRuntimeException();
-        }
-    }
-
-    @Override
-    public void setRowKind(RowKind rowKind) {
-        throw Status.INTERNAL.withDescription("Can't support set row kind").asRuntimeException();
-    }
-
-    @Override
-    public boolean isNullAt(int i) {
-        return sinkRow.get(i) == null;
-    }
-
-    @Override
-    public boolean getBoolean(int i) {
-        return (boolean) sinkRow.get(i);
-    }
-
-    @Override
-    public byte getByte(int i) {
-        return (byte) sinkRow.get(i);
-    }
-
-    @Override
-    public short getShort(int i) {
-        return (short) sinkRow.get(i);
-    }
-
-    @Override
-    public int getInt(int i) {
-        Object value = sinkRow.get(i);
-        if (value instanceof LocalDate) {
-            return (int) ((LocalDate) value).toEpochDay();
-        } else if (value instanceof LocalTime) {
-            // number of milliseconds of the day
-            return (int) (((LocalTime) value).toNanoOfDay() / 1_000_000L);
-        } else {
-            return (int) value;
-        }
-    }
-
-    @Override
-    public long getLong(int i) {
-        return (long) sinkRow.get(i);
-    }
-
-    @Override
-    public float getFloat(int i) {
-        return (float) sinkRow.get(i);
-    }
-
-    @Override
-    public double getDouble(int i) {
-        return (double) sinkRow.get(i);
-    }
-
-    @Override
-    public StringData getString(int i) {
-        return StringData.fromString((String) sinkRow.get(i));
-    }
-
-    @Override
-    public DecimalData getDecimal(int i, int i1, int i2) {
-        return DecimalData.fromBigDecimal((java.math.BigDecimal) sinkRow.get(i), i1, i2);
-    }
-
-    @Override
-    public TimestampData getTimestamp(int i, int i1) {
-        Object value = sinkRow.get(i);
-        if (value instanceof LocalDateTime) {
-            return TimestampData.fromLocalDateTime((LocalDateTime) value);
-        } else if (value instanceof OffsetDateTime) {
-            return TimestampData.fromInstant(((OffsetDateTime) value).toInstant());
-        } else {
-            throw Status.INTERNAL.withDescription("unreachable").asRuntimeException();
-        }
-    }
-
-    @Override
-    public <T> RawValueData<T> getRawValue(int i) {
-        throw Status.INTERNAL
-                .withDescription("Raw type is not supported yet\"")
-                .asRuntimeException();
-    }
-
-    @Override
-    public byte[] getBinary(int i) {
-        return Base64.getDecoder().decode((String) sinkRow.get(i));
-    }
-
-    @Override
-    public ArrayData getArray(int i) {
-        Object[] array = ((ArrayList<?>) sinkRow.get(i)).toArray();
-        return new GenericArrayData(array);
-    }
-
-    @Override
-    public MapData getMap(int i) {
-        throw Status.INTERNAL.withDescription("Can't support Raw").asRuntimeException();
-    }
-
-    @Override
-    public RowData getRow(int i, int i1) {
-        throw new RuntimeException("STRUCT type is not supported yet");
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/DynamicTableSinkContextImpl.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/DynamicTableSinkContextImpl.java
deleted file mode 100644
index 0c6f991b66ab0..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/DynamicTableSinkContextImpl.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime.context;
-
-import java.util.Optional;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.table.connector.sink.DynamicTableSink;
-import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.logical.LogicalType;
-
-/**
- * It is used when flink creating a sink, which should contain the serialization method
- * corresponding to the datatypes of the flink. But the current sink doesn't use it, so it just
- * throws an exception here.
- */
-public class DynamicTableSinkContextImpl implements DynamicTableSink.Context {
-    @Override
-    public boolean isBounded() {
-        return false;
-    }
-
-    @Override
-    public <T> TypeInformation<T> createTypeInformation(DataType consumedDataType) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public <T> TypeInformation<T> createTypeInformation(LogicalType consumedLogicalType) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public DynamicTableSink.DataStructureConverter createDataStructureConverter(
-            DataType consumedDataType) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public Optional<int[][]> getTargetColumns() {
-        throw new UnsupportedOperationException();
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/MailBoxExecImpl.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/MailBoxExecImpl.java
deleted file mode 100644
index cbd1767ec23a4..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/MailBoxExecImpl.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime.context;
-
-import java.util.Arrays;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import org.apache.flink.api.common.operators.MailboxExecutor;
-import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.util.function.ThrowingRunnable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Simple Implementation of Flink MailboxExecutor. Only the mailbox method of executing a mail via
- * `yield` call is implemented. In flink, there is also a background thread that polls for
- * execution, which is not implemented because it is not currently used.
- *
- * <p>Specifically, the class has a queue of different mails saved in the queue, and passed
- * downstream, when the downstream call yield, back to the fifo way to pick the mail execution
- */
-public class MailBoxExecImpl implements MailboxExecutor {
-    // The size of queue is Integer.MAX. In other words, we won't limit the number of emails.
-    BlockingQueue<Mail> queue = new LinkedBlockingQueue<>();
-
-    private static final Logger LOG = LoggerFactory.getLogger(MailBoxExecImpl.class);
-
-    class Mail {
-        ThrowingRunnable<? extends Exception> command;
-        String descriptionFormat;
-        Object[] descriptionArgs;
-
-        public Mail(
-                ThrowingRunnable<? extends Exception> command,
-                String descriptionFormat,
-                Object... descriptionArgs) {
-            this.command = command;
-            this.descriptionFormat = descriptionFormat;
-            this.descriptionArgs = Arrays.stream(descriptionArgs).toArray();
-        }
-    }
-
-    /**
-     * Create a mail and add it to the queue
-     *
-     * @param command the runnable task to add to the mailbox for execution.
-     * @param descriptionFormat the optional description for the command that is used for debugging
-     *     and error-reporting.
-     * @param descriptionArgs the parameters used to format the final description string.
-     */
-    @Override
-    public void execute(
-            ThrowingRunnable<? extends Exception> command,
-            String descriptionFormat,
-            Object... descriptionArgs) {
-        queue.add(new Mail(command, descriptionFormat, descriptionArgs));
-    }
-
-    /** Pick a mail to execute, join the queue as empty, then block and wait */
-    @Override
-    public void yield() throws InterruptedException, FlinkRuntimeException {
-        Mail headMail;
-        while (true) {
-            if ((headMail = queue.poll(1, TimeUnit.SECONDS)) != null) {
-                break;
-            } else {
-                LOG.warn("Query mail queue wait timeout, will continue to wait");
-            }
-        }
-        try {
-            headMail.command.run();
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    /** Pick a mail to execute, join the queue as empty, then return false. */
-    @Override
-    public boolean tryYield() throws FlinkRuntimeException {
-        Mail headMail = queue.poll();
-        if (headMail != null) {
-            try {
-                headMail.command.run();
-            } catch (Exception e) {
-                throw new RuntimeException(e);
-            }
-            return true;
-        } else {
-            return false;
-        }
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/SinkWriterContext.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/SinkWriterContext.java
deleted file mode 100644
index da4d10e97d8bd..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/SinkWriterContext.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime.context;
-
-import java.io.IOException;
-import java.util.OptionalLong;
-import org.apache.flink.api.common.operators.MailboxExecutor;
-import org.apache.flink.api.common.operators.ProcessingTimeService;
-import org.apache.flink.api.connector.sink.Sink;
-import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
-import org.apache.flink.util.UserCodeClassLoader;
-
-/*
- * A simulated implementation of the InitContext, passed down and utilized in the downstream of the sink (SinkV1).
- * Method is implemented in `SinkWriterContextV2`.
- */
-public class SinkWriterContext implements org.apache.flink.api.connector.sink.Sink.InitContext {
-    static class ProcessingTimeCallbackAdapter
-            implements ProcessingTimeService.ProcessingTimeCallback {
-
-        Sink.ProcessingTimeService.ProcessingTimeCallback processingTimerCallback;
-
-        public ProcessingTimeCallbackAdapter(
-                Sink.ProcessingTimeService.ProcessingTimeCallback processingTimerCallback) {
-            this.processingTimerCallback = processingTimerCallback;
-        }
-
-        @Override
-        public void onProcessingTime(long time) throws IOException, InterruptedException {
-            processingTimerCallback.onProcessingTime(time);
-        }
-    }
-
-    /**
-     * Timing task wrapper, responsible for converting `ProcessingTimeService` in sink v2 to
-     * `ProcessingTimeService` in sink v1
-     */
-    static class ProcessingTimeServiceAdapter implements Sink.ProcessingTimeService {
-
-        org.apache.flink.api.common.operators.ProcessingTimeService processingTimeService;
-
-        public ProcessingTimeServiceAdapter(
-                org.apache.flink.api.common.operators.ProcessingTimeService processingTimeService) {
-            this.processingTimeService = processingTimeService;
-        }
-
-        @Override
-        public long getCurrentProcessingTime() {
-            return processingTimeService.getCurrentProcessingTime();
-        }
-
-        @Override
-        public void registerProcessingTimer(
-                long time, ProcessingTimeCallback processingTimerCallback) {
-            processingTimeService.registerTimer(
-                    time, new ProcessingTimeCallbackAdapter(processingTimerCallback));
-        }
-    }
-
-    SinkWriterContextV2 sinkWriterContextV2;
-
-    public SinkWriterContext() {
-        sinkWriterContextV2 = new SinkWriterContextV2();
-    }
-
-    @Override
-    public UserCodeClassLoader getUserCodeClassLoader() {
-        return sinkWriterContextV2.getUserCodeClassLoader();
-    }
-
-    @Override
-    public MailboxExecutor getMailboxExecutor() {
-        return sinkWriterContextV2.getMailboxExecutor();
-    }
-
-    @Override
-    public org.apache.flink.api.connector.sink.Sink.ProcessingTimeService
-            getProcessingTimeService() {
-        return new ProcessingTimeServiceAdapter(sinkWriterContextV2.getProcessingTimeService());
-    }
-
-    @Override
-    public int getSubtaskId() {
-        return sinkWriterContextV2.getSubtaskId();
-    }
-
-    @Override
-    public int getNumberOfParallelSubtasks() {
-        return sinkWriterContextV2.getNumberOfParallelSubtasks();
-    }
-
-    @Override
-    public SinkWriterMetricGroup metricGroup() {
-        return sinkWriterContextV2.metricGroup();
-    }
-
-    @Override
-    public OptionalLong getRestoredCheckpointId() {
-        return sinkWriterContextV2.getRestoredCheckpointId();
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/SinkWriterContextV2.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/SinkWriterContextV2.java
deleted file mode 100644
index 8a89338423c07..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/SinkWriterContextV2.java
+++ /dev/null
@@ -1,181 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime.context;
-
-import java.util.OptionalLong;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.operators.MailboxExecutor;
-import org.apache.flink.api.common.operators.ProcessingTimeService;
-import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.connector.sink2.Sink;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
-import org.apache.flink.util.SimpleUserCodeClassLoader;
-import org.apache.flink.util.UserCodeClassLoader;
-
-/*
- * A simulated implementation of the InitContext, passed down and utilized in the downstream of the sink (SinkV2).
- */
-public class SinkWriterContextV2 implements Sink.InitContext {
-
-    /** A character timer. */
-    class ProcessingTimeServiceImpl implements ProcessingTimeService {
-        private final ScheduledThreadPoolExecutor timerService;
-
-        ProcessingTimeServiceImpl() {
-            this.timerService = new ScheduledThreadPoolExecutor(1);
-        }
-
-        @Override
-        public long getCurrentProcessingTime() {
-            return System.currentTimeMillis();
-        }
-
-        /**
-         * @param timestamp Time when the task is to be executed (in processing time)
-         * @param target The task to be executed
-         * @return
-         */
-        @Override
-        public ScheduledFuture<?> registerTimer(long timestamp, ProcessingTimeCallback target) {
-            long delay = 0;
-            if (timestamp >= getCurrentProcessingTime()) {
-                delay = timestamp - getCurrentProcessingTime() + 1;
-            }
-            return timerService.schedule(
-                    new Runnable() {
-                        @Override
-                        public void run() {
-                            try {
-                                target.onProcessingTime(timestamp);
-                            } catch (Exception e) {
-                                throw new RuntimeException("Callback error" + e.getMessage());
-                            }
-                        }
-                    },
-                    delay,
-                    TimeUnit.MILLISECONDS);
-        }
-    }
-
-    // The sinkWriter's monitor, which is responsible for monitoring downstream data for errors,
-    // currently just mocks one and doesn't do anything with the error
-    SinkWriterMetricGroup sinkWriterMetricGroup;
-    // A character timer.
-    ProcessingTimeService processingTimeService;
-    SimpleUserCodeClassLoader simpleUserCodeClassLoader;
-    // Downstream executes some tasks serially by registering them with `MailboxExecutor`
-    MailboxExecutor mailboxExecutor;
-
-    InitializationContextImpl initializationContext;
-
-    public SinkWriterContextV2() {
-        sinkWriterMetricGroup = new SinkWriterMetircGroupImpl();
-        processingTimeService = new ProcessingTimeServiceImpl();
-        simpleUserCodeClassLoader =
-                SimpleUserCodeClassLoader.create(Thread.currentThread().getContextClassLoader());
-        mailboxExecutor = new MailBoxExecImpl();
-        initializationContext =
-                new InitializationContextImpl(sinkWriterMetricGroup, simpleUserCodeClassLoader);
-    }
-
-    @Override
-    public UserCodeClassLoader getUserCodeClassLoader() {
-        return simpleUserCodeClassLoader;
-    }
-
-    @Override
-    public MailboxExecutor getMailboxExecutor() {
-        return mailboxExecutor;
-    }
-
-    @Override
-    public ProcessingTimeService getProcessingTimeService() {
-        return processingTimeService;
-    }
-
-    @Override
-    public int getSubtaskId() {
-        return (int) (Math.random() * 1000000);
-    }
-
-    @Override
-    public int getNumberOfParallelSubtasks() {
-        return 0;
-    }
-
-    @Override
-    public int getAttemptNumber() {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public SinkWriterMetricGroup metricGroup() {
-        return sinkWriterMetricGroup;
-    }
-
-    @Override
-    public OptionalLong getRestoredCheckpointId() {
-        return OptionalLong.empty();
-    }
-
-    @Override
-    public SerializationSchema.InitializationContext asSerializationSchemaInitializationContext() {
-        return initializationContext;
-    }
-
-    @Override
-    public boolean isObjectReuseEnabled() {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public <IN> TypeSerializer<IN> createInputSerializer() {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public JobID getJobId() {
-        throw new UnsupportedOperationException();
-    }
-
-    class InitializationContextImpl implements SerializationSchema.InitializationContext {
-
-        MetricGroup metricGroup;
-        UserCodeClassLoader userCodeClassLoader;
-
-        InitializationContextImpl(
-                MetricGroup metricGroup, UserCodeClassLoader userCodeClassLoader) {
-            this.metricGroup = metricGroup;
-            this.userCodeClassLoader = userCodeClassLoader;
-        }
-
-        @Override
-        public MetricGroup getMetricGroup() {
-            return metricGroup;
-        }
-
-        @Override
-        public UserCodeClassLoader getUserCodeClassLoader() {
-            return userCodeClassLoader;
-        }
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/SinkWriterMetircGroupImpl.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/SinkWriterMetircGroupImpl.java
deleted file mode 100644
index f7349071d90e4..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/context/SinkWriterMetircGroupImpl.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime.context;
-
-import org.apache.flink.metrics.*;
-import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
-import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
-import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
-
-/**
- * Error monitoring for sink is a mock implementation of SinkWriterMetricGroup, which just simulates
- * the relevant methods and doesn't do anything about the error
- */
-public class SinkWriterMetircGroupImpl extends UnregisteredMetricsGroup
-        implements SinkWriterMetricGroup {
-
-    private final Counter numRecordsOutErrors;
-    private final Counter numRecordsSendErrors;
-    private final Counter numRecordsWritten;
-    private final Counter numBytesWritten;
-    private final OperatorIOMetricGroup operatorIOMetricGroup;
-
-    public SinkWriterMetircGroupImpl() {
-        numRecordsOutErrors = super.counter("numRecordsOutErrors");
-        numRecordsSendErrors = super.counter("numRecordsOutErrors");
-        numRecordsWritten = super.counter("numRecordsOutErrors");
-        numBytesWritten = super.counter("numRecordsOutErrors");
-        operatorIOMetricGroup = UnregisteredMetricsGroup.createOperatorIOMetricGroup();
-    }
-
-    @Override
-    public Counter getNumRecordsOutErrorsCounter() {
-        return numRecordsOutErrors;
-    }
-
-    @Override
-    public Counter getNumRecordsSendErrorsCounter() {
-        return numRecordsSendErrors;
-    }
-
-    @Override
-    public Counter getNumRecordsSendCounter() {
-        return numRecordsWritten;
-    }
-
-    @Override
-    public Counter getNumBytesSendCounter() {
-        return numBytesWritten;
-    }
-
-    @Override
-    public void setCurrentSendTimeGauge(Gauge<Long> currentSendTimeGauge) {
-        super.gauge("currentSendTime", currentSendTimeGauge);
-    }
-
-    @Override
-    public OperatorIOMetricGroup getIOMetricGroup() {
-        return operatorIOMetricGroup;
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/AsyncSinkWriterImpl.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/AsyncSinkWriterImpl.java
deleted file mode 100644
index 6b3a1c2f63f99..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/AsyncSinkWriterImpl.java
+++ /dev/null
@@ -1,126 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime.sinkwriter;
-
-import com.risingwave.connector.api.TableSchema;
-import com.risingwave.connector.api.sink.SinkRow;
-import com.risingwave.mock.flink.runtime.RowDataImpl;
-import com.risingwave.mock.flink.runtime.context.SinkWriterContextV2;
-import com.risingwave.proto.ConnectorServiceProto;
-import java.io.IOException;
-import java.util.Optional;
-import org.apache.flink.api.connector.sink2.SinkWriter;
-import org.apache.flink.table.data.RowData;
-
-/**
- * AsyncSinkWriter is the special case of stateful sink, which is also the only stateful support we
- * currently provide. It involves writing data to the sink by closing the async writer and then
- * re-creating a writer for subsequent writing. This is different from the usage in Flink.
- */
-public class AsyncSinkWriterImpl implements com.risingwave.connector.api.sink.SinkWriter {
-    final org.apache.flink.api.connector.sink2.Sink<RowData> sink;
-
-    SinkWriter<RowData> writer;
-    final TableSchema tableSchema;
-
-    public AsyncSinkWriterImpl(
-            TableSchema tableSchema, org.apache.flink.api.connector.sink2.Sink<RowData> sink) {
-        this.tableSchema = tableSchema;
-        try {
-            this.sink = sink;
-            this.writer = sink.createWriter(new SinkWriterContextV2());
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    private void writeRow(RowData rowData) {
-        try {
-            if (this.writer == null) {
-                this.writer = sink.createWriter(new SinkWriterContextV2());
-            }
-            this.writer.write(
-                    rowData,
-                    new SinkWriter.Context() {
-                        @Override
-                        public long currentWatermark() {
-                            return 0;
-                        }
-
-                        @Override
-                        public Long timestamp() {
-                            return null;
-                        }
-                    });
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    @Override
-    public void beginEpoch(long epoch) {}
-
-    @Override
-    public boolean write(Iterable<SinkRow> rows) {
-        for (SinkRow row : rows) {
-            writeRow(new RowDataImpl(row, tableSchema.getNumColumns()));
-        }
-        return true;
-    }
-
-    @Override
-    public Optional<ConnectorServiceProto.SinkMetadata> barrier(boolean isCheckpoint) {
-        if (isCheckpoint) {
-            try {
-                if (writer != null) {
-                    this.writer.flush(true);
-                    // The reason we need to close writer here is that:
-                    // Flink's async doesn't provide a commit method in the same sense as RW, so
-                    // here we are
-                    // using close writer for commit-like behavior, but closing writer will cause
-                    // the old
-                    // writer to no longer be able to write, so we need to close it here and
-                    // recreate it the
-                    // next time we need to write the data
-                    closeWriter();
-                }
-            } catch (Exception e) {
-                throw new RuntimeException(e);
-            }
-        }
-        return Optional.empty();
-    }
-
-    @Override
-    public void drop() {
-        closeWriter();
-    }
-
-    private void closeWriter() {
-        if (writer == null) {
-            return;
-        }
-        try {
-            writer.close();
-            this.writer = null;
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/CommitRequestImpl.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/CommitRequestImpl.java
deleted file mode 100644
index 89644790e2174..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/CommitRequestImpl.java
+++ /dev/null
@@ -1,64 +0,0 @@
-// Copyright 2024 RisingWave Labs
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package com.risingwave.mock.flink.runtime.sinkwriter;
-
-import org.apache.flink.api.connector.sink2.Committer;
-
-/**
- * The mock implementation of CommitRequest. Rw don't support two-stage committed, so there is no
- * need to simply return Committable
- */
-public class CommitRequestImpl<CommT> implements Committer.CommitRequest<CommT> {
-    CommT committable;
-
-    public CommitRequestImpl(CommT committable) {
-        this.committable = committable;
-    }
-
-    @Override
-    public CommT getCommittable() {
-        return committable;
-    }
-
-    @Override
-    public int getNumberOfRetries() {
-        return 0;
-    }
-
-    @Override
-    public void signalFailedWithKnownReason(Throwable t) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void signalFailedWithUnknownReason(Throwable t) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void retryLater() {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void updateAndRetryLater(Object committable) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void signalAlreadyCommitted() {
-        throw new UnsupportedOperationException();
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/SinkWriterImpl.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/SinkWriterImpl.java
deleted file mode 100644
index e726413a7069e..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/SinkWriterImpl.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime.sinkwriter;
-
-import com.risingwave.connector.api.TableSchema;
-import com.risingwave.connector.api.sink.SinkRow;
-import com.risingwave.mock.flink.runtime.RowDataImpl;
-import com.risingwave.mock.flink.runtime.context.SinkWriterContext;
-import com.risingwave.proto.ConnectorServiceProto;
-import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
-import java.util.Optional;
-import org.apache.flink.api.connector.sink.Committer;
-import org.apache.flink.api.connector.sink.Sink;
-import org.apache.flink.table.data.RowData;
-
-/*
- * The Sink-type sink will invoke this class to write RW data to the downstream.
- */
-public class SinkWriterImpl<CommT> implements com.risingwave.connector.api.sink.SinkWriter {
-    final Sink<RowData, CommT, ?, ?> sink;
-    final org.apache.flink.api.connector.sink.SinkWriter<RowData, CommT, ?> writer;
-
-    final Optional<? extends Committer<CommT>> committer;
-
-    final TableSchema tableSchema;
-
-    public SinkWriterImpl(TableSchema tableSchema, Sink<RowData, CommT, ?, ?> sink) {
-        try {
-            this.tableSchema = tableSchema;
-            this.sink = sink;
-            this.writer = sink.createWriter(new SinkWriterContext(), Collections.emptyList());
-            this.committer = sink.createCommitter();
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    private void writeRow(RowData rowData) {
-        try {
-            writer.write(
-                    rowData,
-                    new org.apache.flink.api.connector.sink.SinkWriter.Context() {
-                        @Override
-                        public long currentWatermark() {
-                            return 0;
-                        }
-
-                        @Override
-                        public Long timestamp() {
-                            return null;
-                        }
-                    });
-        } catch (IOException | InterruptedException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    @Override
-    public void beginEpoch(long epoch) {}
-
-    @Override
-    public boolean write(Iterable<SinkRow> rows) {
-        for (SinkRow row : rows) {
-            writeRow(new RowDataImpl(row, tableSchema.getNumColumns()));
-        }
-        return true;
-    }
-
-    @Override
-    public Optional<ConnectorServiceProto.SinkMetadata> barrier(boolean isCheckpoint) {
-        if (isCheckpoint) {
-            try {
-                List<CommT> objects = writer.prepareCommit(true);
-                if (committer.isPresent()) {
-                    committer.get().commit(objects);
-                } else if (!objects.isEmpty()) {
-                    throw new RuntimeException("Flink sink v1 committer is null");
-                }
-            } catch (Exception e) {
-                throw new RuntimeException(e);
-            }
-        }
-        return Optional.empty();
-    }
-
-    @Override
-    public void drop() {
-        try {
-            writer.close();
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/SinkWriterV2Impl.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/SinkWriterV2Impl.java
deleted file mode 100644
index 4e04115170db8..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/sinkwriter/SinkWriterV2Impl.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * Copyright 2024 RisingWave Labs
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.risingwave.mock.flink.runtime.sinkwriter;
-
-import com.risingwave.connector.api.TableSchema;
-import com.risingwave.connector.api.sink.SinkRow;
-import com.risingwave.mock.flink.runtime.RowDataImpl;
-import com.risingwave.mock.flink.runtime.context.SinkWriterContextV2;
-import com.risingwave.proto.ConnectorServiceProto;
-import java.io.IOException;
-import java.util.Collection;
-import java.util.List;
-import java.util.Optional;
-import java.util.stream.Collectors;
-import org.apache.flink.api.connector.sink2.Committer;
-import org.apache.flink.api.connector.sink2.SinkWriter;
-import org.apache.flink.api.connector.sink2.StatefulSink;
-import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
-import org.apache.flink.table.data.RowData;
-
-/*
- * The SinkV2-type sink will invoke this class to write RW data to the downstream.
- * Stateful sink is currently not supported.
- */
-public class SinkWriterV2Impl<CommT> implements com.risingwave.connector.api.sink.SinkWriter {
-
-    final org.apache.flink.api.connector.sink2.Sink<RowData> sink;
-
-    final SinkWriter<RowData> writer;
-
-    final Optional<Committer<CommT>> committer;
-
-    final TableSchema tableSchema;
-
-    public SinkWriterV2Impl(
-            TableSchema tableSchema, org.apache.flink.api.connector.sink2.Sink<RowData> sink) {
-        try {
-            this.tableSchema = tableSchema;
-            this.sink = sink;
-            this.writer = sink.createWriter(new SinkWriterContextV2());
-            if (writer instanceof StatefulSink.StatefulSinkWriter) {
-                throw new UnsupportedOperationException("Don't support StatefulSink");
-            }
-            if (this.sink instanceof TwoPhaseCommittingSink) {
-                this.committer =
-                        Optional.ofNullable(
-                                ((TwoPhaseCommittingSink<?, CommT>) this.sink).createCommitter());
-            } else {
-                this.committer = Optional.empty();
-            }
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    private void writeRow(RowData rowData) {
-        try {
-            this.writer.write(
-                    rowData,
-                    new SinkWriter.Context() {
-                        @Override
-                        public long currentWatermark() {
-                            return 0;
-                        }
-
-                        @Override
-                        public Long timestamp() {
-                            return null;
-                        }
-                    });
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    @Override
-    public void beginEpoch(long epoch) {}
-
-    @Override
-    public boolean write(Iterable<SinkRow> rows) {
-        for (SinkRow row : rows) {
-            writeRow(new RowDataImpl(row, tableSchema.getNumColumns()));
-        }
-        return true;
-    }
-
-    @Override
-    public Optional<ConnectorServiceProto.SinkMetadata> barrier(boolean isCheckpoint) {
-        if (isCheckpoint) {
-            try {
-                this.writer.flush(false);
-                Collection<CommT> objects =
-                        ((TwoPhaseCommittingSink.PrecommittingSinkWriter<?, CommT>) this.writer)
-                                .prepareCommit();
-                if (committer.isPresent()) {
-                    List<Committer.CommitRequest<CommT>> collect =
-                            objects.stream()
-                                    .map(
-                                            (a) -> {
-                                                return (Committer.CommitRequest<CommT>)
-                                                        new CommitRequestImpl<>(a);
-                                            })
-                                    .collect(Collectors.toList());
-                    committer.get().commit(collect);
-                } else if (!objects.isEmpty()) {
-                    throw new RuntimeException(
-                            "We need to commit Committable, But Committer is null");
-                }
-            } catch (Exception e) {
-                throw new RuntimeException(e);
-            }
-        }
-        return Optional.empty();
-    }
-
-    @Override
-    public void drop() {
-        closeWriter();
-    }
-
-    private void closeWriter() {
-        try {
-            writer.close();
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-}
diff --git a/java/pom.xml b/java/pom.xml
index 6f3ade3f36028..c717cfc87a2e0 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -53,7 +53,6 @@
         <module>connector-node/risingwave-source-cdc-test</module>
         <module>connector-node/assembly</module>
         <module>connector-node/s3-common</module>
-        <module>connector-node/risingwave-sink-mock-flink</module>
         <module>connector-node/tracing</module>
     </modules>
 
@@ -283,21 +282,6 @@
                 <artifactId>risingwave-sink-cassandra</artifactId>
                 <version>${module.version}</version>
             </dependency>
-            <dependency>
-                <groupId>com.risingwave</groupId>
-                <artifactId>risingwave-sink-mock-flink-runtime</artifactId>
-                <version>${module.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>com.risingwave</groupId>
-                <artifactId>risingwave-sink-mock-flink-http-sink</artifactId>
-                <version>${module.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>com.risingwave</groupId>
-                <artifactId>risingwave-sink-mock-flink-common</artifactId>
-                <version>${module.version}</version>
-            </dependency>
             <dependency>
                 <groupId>com.risingwave</groupId>
                 <artifactId>risingwave-sink-jdbc</artifactId>


Test patch:
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/src/test/java/com/risingwave/connector/sink/http/HttpTest.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/src/test/java/com/risingwave/connector/sink/http/HttpTest.java
deleted file mode 100644
index bba264d4d80db..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/src/test/java/com/risingwave/connector/sink/http/HttpTest.java
+++ /dev/null
@@ -1,162 +0,0 @@
-// Copyright 2024 RisingWave Labs
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package com.risingwave.connector.sink.http;
-
-import static java.lang.Thread.sleep;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.Lists;
-import com.risingwave.connector.api.ColumnDesc;
-import com.risingwave.connector.api.TableSchema;
-import com.risingwave.connector.api.sink.ArraySinkRow;
-import com.risingwave.connector.api.sink.SinkRow;
-import com.risingwave.connector.api.sink.SinkWriter;
-import com.risingwave.mock.flink.http.HttpFlinkMockSinkFactory;
-import com.risingwave.mock.flink.runtime.FlinkDynamicAdapterFactory;
-import com.risingwave.proto.Data;
-import com.sun.net.httpserver.HttpExchange;
-import com.sun.net.httpserver.HttpHandler;
-import com.sun.net.httpserver.HttpServer;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.net.InetSocketAddress;
-import java.util.*;
-import java.util.concurrent.Executors;
-import org.junit.Test;
-
-public class HttpTest {
-    MyHttpHandler mockHttpClient() throws IOException {
-        HttpServer httpServer = HttpServer.create(new InetSocketAddress(8083), 0);
-
-        MyHttpHandler myHttpHandler = new MyHttpHandler();
-
-        httpServer.createContext("/test", myHttpHandler);
-
-        httpServer.setExecutor(Executors.newFixedThreadPool(1));
-
-        new Runnable() {
-            @Override
-            public void run() {
-                httpServer.start();
-            }
-        }.run();
-        return myHttpHandler;
-    }
-
-    class MyHttpHandler implements HttpHandler {
-        String resultString;
-        volatile boolean flag;
-
-        public MyHttpHandler() {
-            flag = false;
-        }
-
-        public void restartFlag() {
-            this.flag = false;
-        }
-
-        public void waitHandle() throws InterruptedException {
-            while (!flag) {
-                sleep(500);
-            }
-        }
-
-        @Override
-        public void handle(HttpExchange httpExchange) throws IOException {
-            Scanner scanner =
-                    new Scanner(httpExchange.getRequestBody(), "UTF-8").useDelimiter("\\A");
-            String resultString = scanner.hasNext() ? scanner.next() : "";
-
-            httpExchange.sendResponseHeaders(200, 0);
-            OutputStream outputStream = httpExchange.getResponseBody();
-            String response = "Ok!";
-            outputStream.write(response.getBytes());
-            outputStream.close();
-
-            this.resultString = resultString;
-            this.flag = true;
-        }
-
-        public String getResultString() {
-            return resultString;
-        }
-    }
-
-    void checkList(String resultString, ArrayList<SinkRow> arrayList, TableSchema tableSchema)
-            throws JsonProcessingException {
-        List<ColumnDesc> columnDescs = tableSchema.getColumnDescs();
-
-        for (int i = 0; i < arrayList.size(); i++) {
-            SinkRow sinkRow = arrayList.get(i);
-            TreeMap<String, Object> map1 = new TreeMap<>();
-            for (int j = 0; j < sinkRow.size(); j++) {
-                map1.put(columnDescs.get(j).getName(), sinkRow.get(j));
-            }
-            ObjectMapper objectMapper = new ObjectMapper();
-            String s = objectMapper.writeValueAsString(map1);
-            assert (resultString.contains(s));
-        }
-    }
-
-    @Test
-    public void testHttpSinkWrite() throws IOException, InterruptedException {
-        MyHttpHandler httpHandler = mockHttpClient();
-        TableSchema tableSchema =
-                new TableSchema(
-                        Lists.newArrayList("id", "name"),
-                        Lists.newArrayList(
-                                Data.DataType.newBuilder()
-                                        .setTypeName(Data.DataType.TypeName.INT32)
-                                        .build(),
-                                Data.DataType.newBuilder()
-                                        .setTypeName(Data.DataType.TypeName.VARCHAR)
-                                        .build()),
-                        Lists.newArrayList("id", "name"));
-
-        FlinkDynamicAdapterFactory flinkDynamicAdapterFactory =
-                new FlinkDynamicAdapterFactory(new HttpFlinkMockSinkFactory());
-        HashMap<String, String> config = new HashMap<>();
-        config.put("connector", "http");
-        config.put("url", "http://localhost:8083/test");
-        config.put("format", "json");
-        config.put("type", "append-only");
-        config.put("force_append_only", "true");
-        config.put("primary_key", "id");
-
-        SinkWriter writer = flinkDynamicAdapterFactory.createWriter(tableSchema, config);
-
-        ArrayList<SinkRow> sinkRows = new ArrayList<>();
-        sinkRows.add(new ArraySinkRow(Data.Op.INSERT, 1, "Alice"));
-        sinkRows.add(new ArraySinkRow(Data.Op.INSERT, 2, "xxx"));
-
-        writer.write(sinkRows);
-        writer.barrier(true);
-        httpHandler.waitHandle();
-        httpHandler.restartFlag();
-
-        checkList(httpHandler.getResultString(), sinkRows, tableSchema);
-
-        ArrayList<SinkRow> sinkRows2 = new ArrayList<>();
-        sinkRows2.add(new ArraySinkRow(Data.Op.INSERT, 3, "aaa"));
-        sinkRows2.add(new ArraySinkRow(Data.Op.INSERT, 4, "bbb"));
-
-        writer.write(sinkRows2);
-        writer.barrier(true);
-        httpHandler.waitHandle();
-
-        checkList(httpHandler.getResultString(), sinkRows2, tableSchema);
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/src/test/resources/log4j2.properties b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/src/test/resources/log4j2.properties
deleted file mode 100644
index a380d72fc3c2a..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-http-sink/src/test/resources/log4j2.properties
+++ /dev/null
@@ -1,10 +0,0 @@
-rootLogger.level=ERROR
-# declare the appender to use
-appenders=console
-# appender properties
-appender.console.type=Console
-appender.console.name=stdout
-appender.console.layout.type=PatternLayout
-appender.console.layout.pattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p [%t] %c{2}:%L - %m%n
-rootLogger.appenderRefs=console
-rootLogger.appenderRef.console.ref=stdout
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/test/java/com/risingwave/connector/sink/flink/FlinkMockTest.java b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/test/java/com/risingwave/connector/sink/flink/FlinkMockTest.java
deleted file mode 100644
index 0043925a67be0..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/test/java/com/risingwave/connector/sink/flink/FlinkMockTest.java
+++ /dev/null
@@ -1,501 +0,0 @@
-// Copyright 2024 RisingWave Labs
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package com.risingwave.connector.sink.flink;
-
-import static org.junit.Assert.assertEquals;
-
-import com.google.common.collect.Lists;
-import com.risingwave.connector.api.TableSchema;
-import com.risingwave.connector.api.sink.ArraySinkRow;
-import com.risingwave.connector.api.sink.SinkRow;
-import com.risingwave.mock.flink.common.FlinkDynamicAdapterConfig;
-import com.risingwave.mock.flink.common.FlinkMockSinkFactory;
-import com.risingwave.mock.flink.runtime.FlinkDynamicAdapterFactory;
-import com.risingwave.proto.Data;
-import io.grpc.StatusRuntimeException;
-import java.io.IOException;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.OffsetDateTime;
-import java.time.ZoneOffset;
-import java.util.*;
-import org.apache.flink.api.connector.sink.Committer;
-import org.apache.flink.api.connector.sink.GlobalCommitter;
-import org.apache.flink.api.connector.sink.Sink;
-import org.apache.flink.api.connector.sink.SinkWriter;
-import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.flink.table.connector.ChangelogMode;
-import org.apache.flink.table.connector.sink.DynamicTableSink;
-import org.apache.flink.table.connector.sink.SinkProvider;
-import org.apache.flink.table.connector.sink.SinkV2Provider;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.factories.DynamicTableSinkFactory;
-import org.junit.Test;
-
-public class FlinkMockTest {
-    private static final String SINK_V1 = "sinkV1";
-    private static final String SINK_V2 = "sinkV2";
-
-    private static MockStorage mockStorage = new MockStorage();
-
-    class MockFlinkMockSinkFactory implements FlinkMockSinkFactory {
-        final String mockSinkType;
-
-        public MockFlinkMockSinkFactory(String mockSinkType) {
-            this.mockSinkType = mockSinkType;
-        }
-
-        @Override
-        public void validate(TableSchema tableSchema, FlinkDynamicAdapterConfig config)
-                throws StatusRuntimeException {}
-
-        @Override
-        public DynamicTableSinkFactory getDynamicTableSinkFactory() {
-            return new MockDynamicTableSinkFactory(mockSinkType);
-        }
-    }
-
-    class MockDynamicTableSinkFactory implements DynamicTableSinkFactory {
-        final String mockSinkType;
-
-        public MockDynamicTableSinkFactory(String mockSinkType) {
-            this.mockSinkType = mockSinkType;
-        }
-
-        @Override
-        public DynamicTableSink createDynamicTableSink(Context context) {
-            return new MockDynamicTableSink(mockSinkType);
-        }
-
-        @Override
-        public String factoryIdentifier() {
-            throw new UnsupportedOperationException();
-        }
-
-        @Override
-        public Set<ConfigOption<?>> requiredOptions() {
-            return null;
-        }
-
-        @Override
-        public Set<ConfigOption<?>> optionalOptions() {
-            return null;
-        }
-    }
-
-    class MockDynamicTableSink implements DynamicTableSink {
-        final String mockSinkType;
-
-        public MockDynamicTableSink(String mockSinkType) {
-            this.mockSinkType = mockSinkType;
-        }
-
-        @Override
-        public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
-            throw new UnsupportedOperationException();
-        }
-
-        @Override
-        public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
-            if (mockSinkType.equals(SINK_V1)) {
-                return SinkProvider.of(new MockSinkV1());
-            } else if (mockSinkType.equals(SINK_V2)) {
-                return SinkV2Provider.of(new MockSinkV2());
-            } else {
-                throw new RuntimeException("Don't support sink type");
-            }
-        }
-
-        @Override
-        public DynamicTableSink copy() {
-            throw new UnsupportedOperationException();
-        }
-
-        @Override
-        public String asSummaryString() {
-            throw new UnsupportedOperationException();
-        }
-    }
-
-    class MockSinkV1 implements Sink {
-
-        @Override
-        public org.apache.flink.api.connector.sink.SinkWriter createWriter(
-                InitContext context, List states) throws IOException {
-            return new MockSinkWriterV1();
-        }
-
-        @Override
-        public Optional<SimpleVersionedSerializer> getWriterStateSerializer() {
-            throw new UnsupportedOperationException();
-        }
-
-        @Override
-        public Optional<Committer> createCommitter() throws IOException {
-            return Optional.of(new MockSinkCommitterV1());
-        }
-
-        @Override
-        public Optional<GlobalCommitter> createGlobalCommitter() throws IOException {
-            throw new UnsupportedOperationException();
-        }
-
-        @Override
-        public Optional<SimpleVersionedSerializer> getCommittableSerializer() {
-            throw new UnsupportedOperationException();
-        }
-
-        @Override
-        public Optional<SimpleVersionedSerializer> getGlobalCommittableSerializer() {
-            throw new UnsupportedOperationException();
-        }
-    }
-
-    class MockSinkWriterV1 implements SinkWriter {
-        List<RowData> cacheList = new ArrayList<>();
-
-        @Override
-        public void write(Object element, Context context)
-                throws IOException, InterruptedException {
-            cacheList.add((RowData) element);
-        }
-
-        @Override
-        public List prepareCommit(boolean flush) throws IOException, InterruptedException {
-            int id = mockStorage.addData(new ArrayList<>(cacheList));
-            cacheList.clear();
-            return Lists.newArrayList(id);
-        }
-
-        @Override
-        public void close() throws Exception {}
-    }
-
-    class MockSinkCommitterV1 implements Committer {
-
-        public MockSinkCommitterV1() {}
-
-        @Override
-        public List commit(List committables) throws IOException, InterruptedException {
-            mockStorage.commitId(committables);
-            return null;
-        }
-
-        @Override
-        public void close() throws Exception {}
-    }
-
-    class MockSinkV2 implements org.apache.flink.api.connector.sink2.Sink, TwoPhaseCommittingSink {
-
-        @Override
-        public PrecommittingSinkWriter createWriter(InitContext context) throws IOException {
-            return new MockSinkWriterV2();
-        }
-
-        @Override
-        public org.apache.flink.api.connector.sink2.Committer createCommitter() throws IOException {
-            return null;
-        }
-
-        @Override
-        public SimpleVersionedSerializer getCommittableSerializer() {
-            throw new UnsupportedOperationException();
-        }
-    }
-
-    class MockSinkWriterV2
-            implements org.apache.flink.api.connector.sink2.SinkWriter,
-                    TwoPhaseCommittingSink.PrecommittingSinkWriter {
-        List<RowData> cacheList = new ArrayList<>();
-
-        List<Integer> ids = new ArrayList<>();
-
-        @Override
-        public void write(Object element, Context context)
-                throws IOException, InterruptedException {
-            cacheList.add((RowData) element);
-        }
-
-        @Override
-        public void flush(boolean endOfInput) throws IOException, InterruptedException {
-            int id = mockStorage.addData(new ArrayList<>(cacheList));
-            cacheList.clear();
-            ids.add(id);
-        }
-
-        @Override
-        public void close() throws Exception {}
-
-        @Override
-        public Collection prepareCommit() throws IOException, InterruptedException {
-            mockStorage.commitId(ids);
-            return Collections.emptyList();
-        }
-    }
-
-    static class MockStorage {
-        HashMap<Integer, List<RowData>> tempMap;
-
-        int id;
-
-        TreeMap<Integer, List<RowData>> commitMap;
-
-        public MockStorage() {
-            tempMap = new HashMap<>();
-            id = 0;
-            commitMap = new TreeMap<>();
-        }
-
-        public int addData(List<RowData> data) {
-            int id1 = id++;
-            tempMap.put(id1, data);
-            return id1;
-        }
-
-        public void commitId(List<Integer> ids) {
-            for (int id : ids) {
-                List<RowData> data = tempMap.remove(id);
-                commitMap.put(id, data);
-            }
-        }
-    }
-
-    private TableSchema getTableSchema() {
-        return new TableSchema(
-                Lists.newArrayList(
-                        "id", "name", "v_date", "v_time", "v_timestamp", "v_timestamptz"),
-                Lists.newArrayList(
-                        Data.DataType.newBuilder()
-                                .setTypeName(Data.DataType.TypeName.INT32)
-                                .build(),
-                        Data.DataType.newBuilder()
-                                .setTypeName(Data.DataType.TypeName.VARCHAR)
-                                .build(),
-                        Data.DataType.newBuilder().setTypeName(Data.DataType.TypeName.DATE).build(),
-                        Data.DataType.newBuilder().setTypeName(Data.DataType.TypeName.TIME).build(),
-                        Data.DataType.newBuilder()
-                                .setTypeName(Data.DataType.TypeName.TIMESTAMP)
-                                .build(),
-                        Data.DataType.newBuilder()
-                                .setTypeName(Data.DataType.TypeName.TIMESTAMPTZ)
-                                .build()),
-                Lists.newArrayList("id", "name"));
-    }
-
-    @Test
-    public void testSinkWriteV1() {
-        FlinkDynamicAdapterFactory flinkDynamicAdapterFactory =
-                new FlinkDynamicAdapterFactory(new MockFlinkMockSinkFactory(SINK_V1));
-        com.risingwave.connector.api.sink.SinkWriter writer =
-                flinkDynamicAdapterFactory.createWriter(getTableSchema(), new HashMap<>());
-        List<SinkRow> sinkRows =
-                java.util.Arrays.asList(
-                        new ArraySinkRow(
-                                Data.Op.INSERT,
-                                1,
-                                "Alice",
-                                LocalDate.ofEpochDay(0),
-                                LocalTime.of(0, 0, 0, 0),
-                                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000),
-                                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC)),
-                        new ArraySinkRow(
-                                Data.Op.INSERT,
-                                2,
-                                "Bob",
-                                LocalDate.ofEpochDay(0),
-                                LocalTime.of(0, 0, 0, 1000000),
-                                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000),
-                                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC)));
-
-        writer.write(sinkRows);
-        writer.barrier(true);
-
-        RowData rowData1 = mockStorage.commitMap.lastEntry().getValue().get(0);
-        assertEquals(rowData1.getArity(), 6);
-        assertEquals(rowData1.getInt(0), 1);
-        assertEquals(rowData1.getString(1).toString(), "Alice");
-        assertEquals(rowData1.getInt(2), 0);
-        assertEquals(rowData1.getInt(3), 0);
-        assertEquals(
-                rowData1.getTimestamp(4, 0).toLocalDateTime(),
-                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000));
-        assertEquals(
-                rowData1.getTimestamp(5, 0).toInstant(),
-                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC).toInstant());
-
-        RowData rowData2 = mockStorage.commitMap.lastEntry().getValue().get(1);
-        assertEquals(rowData2.getArity(), 6);
-        assertEquals(rowData2.getInt(0), 2);
-        assertEquals(rowData2.getString(1).toString(), "Bob");
-        assertEquals(rowData2.getInt(2), 0);
-        assertEquals(rowData2.getInt(3), 1);
-        assertEquals(
-                rowData2.getTimestamp(4, 0).toLocalDateTime(),
-                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000));
-        assertEquals(
-                rowData2.getTimestamp(5, 0).toInstant(),
-                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC).toInstant());
-
-        List<SinkRow> sinkRows2 =
-                java.util.Arrays.asList(
-                        new ArraySinkRow(
-                                Data.Op.INSERT,
-                                3,
-                                "xxx",
-                                LocalDate.ofEpochDay(0),
-                                LocalTime.of(0, 0, 0, 0),
-                                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000),
-                                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC)),
-                        new ArraySinkRow(
-                                Data.Op.INSERT,
-                                4,
-                                "hhh",
-                                LocalDate.ofEpochDay(0),
-                                LocalTime.of(0, 0, 0, 0),
-                                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000),
-                                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC)));
-
-        writer.write(sinkRows2);
-        writer.barrier(true);
-
-        RowData rowData3 = mockStorage.commitMap.lastEntry().getValue().get(0);
-        assertEquals(rowData3.getArity(), 6);
-        assertEquals(rowData3.getInt(0), 3);
-        assertEquals(rowData3.getString(1).toString(), "xxx");
-        assertEquals(rowData3.getInt(2), 0);
-        assertEquals(rowData3.getInt(3), 0);
-        assertEquals(
-                rowData3.getTimestamp(4, 0).toLocalDateTime(),
-                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000));
-        assertEquals(
-                rowData3.getTimestamp(5, 0).toInstant(),
-                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC).toInstant());
-
-        RowData rowData4 = mockStorage.commitMap.lastEntry().getValue().get(1);
-        assertEquals(rowData4.getArity(), 6);
-        assertEquals(rowData4.getInt(0), 4);
-        assertEquals(rowData4.getString(1).toString(), "hhh");
-        assertEquals(rowData4.getInt(2), 0);
-        assertEquals(rowData4.getInt(3), 0);
-        assertEquals(
-                rowData4.getTimestamp(4, 0).toLocalDateTime(),
-                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000));
-        assertEquals(rowData4.getTimestamp(5, 0).toString(), "1970-01-01T00:00:01.000001");
-    }
-
-    @Test
-    public void testSinkWriteV2() {
-        FlinkDynamicAdapterFactory flinkDynamicAdapterFactory =
-                new FlinkDynamicAdapterFactory(new MockFlinkMockSinkFactory(SINK_V2));
-        com.risingwave.connector.api.sink.SinkWriter writer =
-                flinkDynamicAdapterFactory.createWriter(getTableSchema(), new HashMap<>());
-        List<SinkRow> sinkRows =
-                java.util.Arrays.asList(
-                        new ArraySinkRow(
-                                Data.Op.INSERT,
-                                1,
-                                "Alice",
-                                LocalDate.ofEpochDay(0),
-                                LocalTime.of(0, 0, 0, 0),
-                                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000),
-                                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC)),
-                        new ArraySinkRow(
-                                Data.Op.INSERT,
-                                2,
-                                "Bob",
-                                LocalDate.ofEpochDay(0),
-                                LocalTime.of(0, 0, 0, 1000000),
-                                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000),
-                                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC)));
-
-        writer.write(sinkRows);
-        writer.barrier(true);
-
-        RowData rowData1 = mockStorage.commitMap.lastEntry().getValue().get(0);
-        assertEquals(rowData1.getArity(), 6);
-        assertEquals(rowData1.getInt(0), 1);
-        assertEquals(rowData1.getString(1).toString(), "Alice");
-        assertEquals(rowData1.getInt(2), 0);
-        assertEquals(rowData1.getInt(3), 0);
-        assertEquals(
-                rowData1.getTimestamp(4, 0).toLocalDateTime(),
-                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000));
-        assertEquals(
-                rowData1.getTimestamp(5, 0).toInstant(),
-                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC).toInstant());
-
-        RowData rowData2 = mockStorage.commitMap.lastEntry().getValue().get(1);
-        assertEquals(rowData2.getArity(), 6);
-        assertEquals(rowData2.getInt(0), 2);
-        assertEquals(rowData2.getString(1).toString(), "Bob");
-        assertEquals(rowData2.getInt(2), 0);
-        assertEquals(rowData2.getInt(3), 1);
-        assertEquals(
-                rowData2.getTimestamp(4, 0).toLocalDateTime(),
-                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000));
-        assertEquals(
-                rowData2.getTimestamp(5, 0).toInstant(),
-                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC).toInstant());
-
-        List<SinkRow> sinkRows2 =
-                java.util.Arrays.asList(
-                        new ArraySinkRow(
-                                Data.Op.INSERT,
-                                3,
-                                "xxx",
-                                LocalDate.ofEpochDay(0),
-                                LocalTime.of(0, 0, 0, 0),
-                                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000),
-                                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC)),
-                        new ArraySinkRow(
-                                Data.Op.INSERT,
-                                4,
-                                "hhh",
-                                LocalDate.ofEpochDay(0),
-                                LocalTime.of(0, 0, 0, 0),
-                                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000),
-                                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC)));
-
-        writer.write(sinkRows2);
-        writer.barrier(true);
-
-        RowData rowData3 = mockStorage.commitMap.lastEntry().getValue().get(0);
-        assertEquals(rowData3.getArity(), 6);
-        assertEquals(rowData3.getInt(0), 3);
-        assertEquals(rowData3.getString(1).toString(), "xxx");
-        assertEquals(rowData3.getInt(2), 0);
-        assertEquals(rowData3.getInt(3), 0);
-        assertEquals(
-                rowData3.getTimestamp(4, 0).toLocalDateTime(),
-                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000));
-        assertEquals(
-                rowData3.getTimestamp(5, 0).toInstant(),
-                OffsetDateTime.of(1970, 1, 1, 0, 0, 1, 1000, ZoneOffset.UTC).toInstant());
-
-        RowData rowData4 = mockStorage.commitMap.lastEntry().getValue().get(1);
-        assertEquals(rowData4.getArity(), 6);
-        assertEquals(rowData4.getInt(0), 4);
-        assertEquals(rowData4.getString(1).toString(), "hhh");
-        assertEquals(rowData4.getInt(2), 0);
-        assertEquals(rowData4.getInt(3), 0);
-        assertEquals(
-                rowData4.getTimestamp(4, 0).toLocalDateTime(),
-                LocalDateTime.of(1970, 1, 1, 0, 0, 0, 1000));
-        assertEquals(rowData4.getTimestamp(5, 0).toString(), "1970-01-01T00:00:01.000001");
-    }
-}
diff --git a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/test/resources/log4j2.properties b/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/test/resources/log4j2.properties
deleted file mode 100644
index a380d72fc3c2a..0000000000000
--- a/java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/test/resources/log4j2.properties
+++ /dev/null
@@ -1,10 +0,0 @@
-rootLogger.level=ERROR
-# declare the appender to use
-appenders=console
-# appender properties
-appender.console.type=Console
-appender.console.name=stdout
-appender.console.layout.type=PatternLayout
-appender.console.layout.pattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p [%t] %c{2}:%L - %m%n
-rootLogger.appenderRefs=console
-rootLogger.appenderRef.console.ref=stdout