chore(sink): exclude java iceberg and deltalake sink from connector node

ci/scripts/connector-node-integration-test.sh

@@ -113,45 +113,6 @@ for ((i=0; i<${#type[@]}; i++)); do
       echo "File sink ${type[i]} test failed"
       exit 1
     fi
-
-    # test upsert mode
-    echo "--- running iceberg upsert mode ${type[i]} integration tests"
-    cd ${RISINGWAVE_ROOT}/java/connector-node/python-client
-    python3 pyspark-util.py create_iceberg
-    if python3 integration_tests.py --upsert_iceberg_sink ${upsert_sink_input_feature[i]}; then
-      python3 pyspark-util.py test_upsert_iceberg --input_file="./data/upsert_sink_input.json"
-      echo "Upsert iceberg sink ${type[i]} test passed"
-    else
-      echo "Upsert iceberg sink ${type[i]} test failed"
-      exit 1
-    fi
-    python3 pyspark-util.py drop_iceberg
-
-    # test append-only mode
-    echo "--- running iceberg append-only mode ${type[i]} integration tests"
-    cd ${RISINGWAVE_ROOT}/java/connector-node/python-client
-    python3 pyspark-util.py create_iceberg
-    if python3 integration_tests.py --iceberg_sink ${sink_input_feature[i]}; then
-      python3 pyspark-util.py test_iceberg
-      echo "Iceberg sink ${type[i]} test passed"
-    else
-      echo "Iceberg sink ${type[i]} test failed"
-      exit 1
-    fi
-    python3 pyspark-util.py drop_iceberg
-
-    # test append-only mode
-    echo "--- running deltalake append-only mod ${type[i]} integration tests"
-    cd ${RISINGWAVE_ROOT}/java/connector-node/python-client
-    python3 pyspark-util.py create_deltalake
-    if python3 integration_tests.py --deltalake_sink ${sink_input_feature[i]}; then
-      python3 pyspark-util.py test_deltalake
-      echo "Deltalake sink ${type[i]} test passed"
-    else
-      echo "Deltalake sink ${type[i]} test failed"
-      exit 1
-    fi
-    python3 pyspark-util.py clean_deltalake
 done
 
 

e2e_test/sink/deltalake_rust_sink.slt

@@ -7,7 +7,7 @@ CREATE MATERIALIZED VIEW mv6 AS SELECT * FROM t6;
 statement ok
 create sink s6 as select * from mv6
 with (
-    connector = 'deltalake_rust',
+    connector = 'deltalake',
     type = 'append-only',
     force_append_only = 'true',
     location = 's3a://deltalake/deltalake-test',

e2e_test/sink/iceberg_sink.slt

@@ -6,15 +6,17 @@ CREATE MATERIALIZED VIEW mv6 AS SELECT * FROM t6;
 
 statement ok
 CREATE SINK s6 AS select mv6.v1 as v1, mv6.v2 as v2, mv6.v3 as v3 from mv6 WITH (
-    connector = 'iceberg_java',
+    connector = 'iceberg',
     type = 'upsert',
     primary_key = 'v1',
     warehouse.path = 's3://iceberg',
     s3.endpoint = 'http://127.0.0.1:9301',
     s3.access.key = 'hummockadmin',
     s3.secret.key = 'hummockadmin',
-    database.name='demo_db',
-    table.name='demo_table'
+    s3.region = 'us-east-1',
+    catalog.type = 'storage',
+    database.name='demo',
+    table.name='demo_db.demo_table'
 );
 
 statement ok

integration_tests/iceberg-sink/create_sink.sql

@@ -1,13 +1,15 @@
 CREATE SINK bhv_iceberg_sink
 FROM
     bhv_mv WITH (
-    connector = 'iceberg_java',
+    connector = 'iceberg',
     type = 'upsert',
     primary_key = 'user_id, target_id, event_timestamp',
     warehouse.path = 's3://hummock001/iceberg-data',
     s3.endpoint = 'http://minio-0:9301',
     s3.access.key = 'hummockadmin',
     s3.secret.key = 'hummockadmin',
-    database.name='demo_db',
-    table.name='demo_table'
+    s3.region = 'us-east-1',
+    catalog.type = 'storage',
+    database.name='demo',
+    table.name='demo_db.demo_table'
 );

java/connector-node/assembly/assembly.xml

@@ -42,11 +42,8 @@
                 <include>*:risingwave-sink-es-7</include>
                 <include>*:risingwave-sink-cassandra</include>
                 <include>*:risingwave-sink-jdbc</include>
-                <include>*:risingwave-sink-iceberg</include>
-                <include>*:risingwave-sink-deltalake</include>
+                <include>*:risingwave-sink-mock-flink-http-sink</include>
 
-                <!-- For S3-->
-                <include>*:s3-common</include>
             </includes>
             <useTransitiveDependencies>true</useTransitiveDependencies>
             <useTransitiveFiltering>true</useTransitiveFiltering>

java/connector-node/risingwave-connector-service/pom.xml

@@ -38,6 +38,11 @@
             <groupId>com.risingwave</groupId>
             <artifactId>connector-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
 
         <dependency>
             <groupId>com.google.code.gson</groupId>
@@ -77,16 +82,6 @@
             <artifactId>risingwave-sink-jdbc</artifactId>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>risingwave-sink-iceberg</artifactId>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>risingwave-sink-deltalake</artifactId>
-            <scope>provided</scope>
-        </dependency>
         <dependency>
             <groupId>com.risingwave</groupId>
             <artifactId>risingwave-sink-es-7</artifactId>
@@ -104,6 +99,7 @@
         <dependency>
             <groupId>com.risingwave</groupId>
             <artifactId>risingwave-sink-mock-flink-http-sink</artifactId>
+            <scope>provided</scope>
         </dependency>
     </dependencies>
 </project>

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkUtils.java

@@ -39,10 +39,6 @@ public static SinkFactory getSinkFactory(String sinkName) {
                 return new FileSinkFactory();
             case "jdbc":
                 return new JDBCSinkFactory();
-            case "iceberg_java":
-                return new IcebergSinkFactory();
-            case "deltalake":
-                return new DeltaLakeSinkFactory();
             case "elasticsearch":
                 return new EsSinkFactory();
             case "cassandra":

java/connector-node/risingwave-sink-es-7/pom.xml

@@ -51,6 +51,10 @@
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>elasticsearch-rest-high-level-client</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

java/pom.xml

@@ -309,6 +309,16 @@
                 <artifactId>gson</artifactId>
                 <version>${gson.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.google.guava</groupId>
+                <artifactId>guava</artifactId>
+                <version>33.0.0-jre</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.httpcomponents</groupId>
+                <artifactId>httpclient</artifactId>
+                <version>4.5.10</version>
+            </dependency>
             <dependency>
                 <groupId>io.prometheus</groupId>
                 <artifactId>simpleclient_httpserver</artifactId>

src/connector/src/sink/deltalake.rs

@@ -46,7 +46,7 @@ use super::{
 };
 use crate::sink::writer::SinkWriterExt;
 
-pub const DELTALAKE_SINK: &str = "deltalake_rust";
+pub const DELTALAKE_SINK: &str = "deltalake";
 pub const DEFAULT_REGION: &str = "us-east-1";
 
 #[derive(Deserialize, Serialize, Debug, Clone, WithOptions)]
@@ -534,7 +534,7 @@ mod test {
             .unwrap();
 
         let properties = hashmap! {
-            "connector".to_string() => "deltalake_rust".to_string(),
+            "connector".to_string() => "deltalake".to_string(),
             "force_append_only".to_string() => "true".to_string(),
             "type".to_string() => "append-only".to_string(),
             "location".to_string() => format!("file://{}", path),

src/connector/src/sink/iceberg/mod.rs

@@ -51,22 +51,11 @@ use super::{
 };
 use crate::deserialize_bool_from_string;
 use crate::sink::coordinate::CoordinatedSinkWriter;
-use crate::sink::remote::{CoordinatedRemoteSink, RemoteSinkTrait};
 use crate::sink::writer::{LogSinkerOf, SinkWriter, SinkWriterExt};
 use crate::sink::{Result, SinkCommitCoordinator, SinkParam};
 
 /// This iceberg sink is WIP. When it ready, we will change this name to "iceberg".
 pub const ICEBERG_SINK: &str = "iceberg";
-pub const REMOTE_ICEBERG_SINK: &str = "iceberg_java";
-
-#[derive(Debug)]
-pub struct RemoteIceberg;
-
-impl RemoteSinkTrait for RemoteIceberg {
-    const SINK_NAME: &'static str = REMOTE_ICEBERG_SINK;
-}
-
-pub type RemoteIcebergSink = CoordinatedRemoteSink<RemoteIceberg>;
 
 #[derive(Debug, Clone, Deserialize, WithOptions)]
 #[serde(deny_unknown_fields)]

src/connector/src/sink/mod.rs

@@ -79,15 +79,13 @@ macro_rules! for_all_sinks {
                 { ClickHouse, $crate::sink::clickhouse::ClickHouseSink },
                 { Iceberg, $crate::sink::iceberg::IcebergSink },
                 { Nats, $crate::sink::nats::NatsSink },
-                { RemoteIceberg, $crate::sink::iceberg::RemoteIcebergSink },
                 { Jdbc, $crate::sink::remote::JdbcSink },
-                { DeltaLake, $crate::sink::remote::DeltaLakeSink },
                 { ElasticSearch, $crate::sink::remote::ElasticSearchSink },
                 { Cassandra, $crate::sink::remote::CassandraSink },
                 { HttpJava, $crate::sink::remote::HttpJavaSink },
                 { Doris, $crate::sink::doris::DorisSink },
                 { Starrocks, $crate::sink::starrocks::StarrocksSink },
-                { DeltaLakeRust, $crate::sink::deltalake::DeltaLakeSink },
+                { DeltaLake, $crate::sink::deltalake::DeltaLakeSink },
                 { BigQuery, $crate::sink::big_query::BigQuerySink },
                 { Test, $crate::sink::test_sink::TestSink },
                 { Table, $crate::sink::table::TableSink }

src/frontend/src/optimizer/plan_node/stream_sink.rs

@@ -34,7 +34,6 @@ use risingwave_connector::sink::{
     SINK_TYPE_UPSERT, SINK_USER_FORCE_APPEND_ONLY_OPTION,
 };
 use risingwave_pb::stream_plan::stream_node::PbNodeBody;
-use tracing::info;
 
 use super::derive::{derive_columns, derive_pk};
 use super::generic::GenericPlanRef;
@@ -150,13 +149,6 @@ impl StreamSink {
             Distribution::Single => RequiredDist::single(),
             _ => {
                 match properties.get("connector") {
-                    Some(s) if s == "deltalake" => {
-                        // iceberg with multiple parallelism will fail easily with concurrent commit
-                        // on metadata
-                        // TODO: reset iceberg sink to have multiple parallelism
-                        info!("setting iceberg sink parallelism to singleton");
-                        RequiredDist::single()
-                    }
                     Some(s) if s == "jdbc" && sink_type == SinkType::Upsert => {
                         if sink_type == SinkType::Upsert && downstream_pk.is_empty() {
                             return Err(ErrorCode::SinkError(Box::new(Error::new(