source: soft-fail with a flag to achive backwards compatibility

Author: xxchan

src/batch/src/executor/source.rs

@@ -63,7 +63,8 @@ impl BoxedExecutorBuilder for SourceExecutor {
         // prepare connector source
         let source_props: HashMap<String, String> =
             HashMap::from_iter(source_node.properties.clone());
-        let config = ConnectorProperties::extract(source_props).map_err(BatchError::connector)?;
+        let config =
+            ConnectorProperties::extract(source_props, false).map_err(BatchError::connector)?;
 
         let info = source_node.get_info().unwrap();
         let parser_config = SpecificParserConfig::new(info, &source_node.properties)?;

src/connector/src/sink/kafka.rs

@@ -263,6 +263,7 @@ impl From<KafkaConfig> for KafkaProperties {
             common: val.common,
             rdkafka_properties_common: val.rdkafka_properties_common,
             rdkafka_properties_consumer: Default::default(),
+            unknown_fields: Default::default(),
         }
     }
 }

src/connector/src/source/base.rs

@@ -55,8 +55,9 @@ const SPLIT_TYPE_FIELD: &str = "split_type";
 const SPLIT_INFO_FIELD: &str = "split_info";
 pub const UPSTREAM_SOURCE_KEY: &str = "connector";
 
-pub trait TryFromHashmap: Sized {
-    fn try_from_hashmap(props: HashMap<String, String>) -> Result<Self>;
+pub trait TryFromHashmap: Sized + UnknownFields {
+    /// Used to initialize the source properties from the raw untyped `WITH` options.
+    fn try_from_hashmap(props: HashMap<String, String>, deny_unknown_fields: bool) -> Result<Self>;
 }
 
 pub trait SourceProperties: TryFromHashmap + Clone {
@@ -68,10 +69,24 @@ pub trait SourceProperties: TryFromHashmap + Clone {
     fn init_from_pb_source(&mut self, _source: &PbSource) {}
 }
 
-impl<P: DeserializeOwned> TryFromHashmap for P {
-    fn try_from_hashmap(props: HashMap<String, String>) -> Result<Self> {
+pub trait UnknownFields {
+    /// Unrecognized fields in the `WITH` clause.
+    fn unknown_fields(&self) -> HashMap<String, String>;
+}
+
+impl<P: DeserializeOwned + UnknownFields> TryFromHashmap for P {
+    fn try_from_hashmap(props: HashMap<String, String>, deny_unknown_fields: bool) -> Result<Self> {
         let json_value = serde_json::to_value(props).map_err(|e| anyhow!(e))?;
-        serde_json::from_value::<P>(json_value).map_err(|e| anyhow!(e.to_string()))
+        let res = serde_json::from_value::<P>(json_value).map_err(|e| anyhow!(e.to_string()))?;
+
+        if !deny_unknown_fields || res.unknown_fields().is_empty() {
+            Ok(res)
+        } else {
+            Err(anyhow!(
+                "Unknown fields in the WITH clause: {:?}",
+                res.unknown_fields()
+            ))
+        }
     }
 }
 
@@ -397,13 +412,16 @@ impl ConnectorProperties {
     /// Creates typed source properties from the raw `WITH` properties.
     ///
     /// It checks the `connector` field, and them dispatches to the corresponding type's `try_from_hashmap` method.
-    pub fn extract(mut props: HashMap<String, String>) -> Result<Self> {
+    ///
+    /// `deny_unknown_fields`: Since `WITH` options are persisted in meta, we do not deny unknown fields when restoring from
+    /// existing data to avoid breaking backwards compatibility. We only deny unknown fields when creating new sources.
+    pub fn extract(mut props: HashMap<String, String>, deny_unknown_fields: bool) -> Result<Self> {
         if Self::is_new_fs_connector_hash_map(&props) {
             _ = props
                 .remove(UPSTREAM_SOURCE_KEY)
                 .ok_or_else(|| anyhow!("Must specify 'connector' in WITH clause"))?;
             return Ok(ConnectorProperties::S3(Box::new(
-                S3Properties::try_from_hashmap(props)?,
+                S3Properties::try_from_hashmap(props, deny_unknown_fields)?,
             )));
         }
 
@@ -413,7 +431,7 @@ impl ConnectorProperties {
         match_source_name_str!(
             connector.to_lowercase().as_str(),
             PropType,
-            PropType::try_from_hashmap(props).map(ConnectorProperties::from),
+            PropType::try_from_hashmap(props, deny_unknown_fields).map(ConnectorProperties::from),
             |other| Err(anyhow!("connector '{}' is not supported", other))
         )
     }
@@ -683,7 +701,7 @@ mod tests {
             "nexmark.split.num" => "1",
         ));
 
-        let props = ConnectorProperties::extract(props).unwrap();
+        let props = ConnectorProperties::extract(props, true).unwrap();
 
         if let ConnectorProperties::Nexmark(props) = props {
             assert_eq!(props.table_type, Some(EventType::Person));
@@ -703,7 +721,7 @@ mod tests {
             "broker.rewrite.endpoints" => r#"{"b-1:9092":"dns-1", "b-2:9092":"dns-2"}"#,
         ));
 
-        let props = ConnectorProperties::extract(props).unwrap();
+        let props = ConnectorProperties::extract(props, true).unwrap();
         if let ConnectorProperties::Kafka(k) = props {
             assert!(k.common.broker_rewrite_map.is_some());
             println!("{:?}", k.common.broker_rewrite_map);
@@ -737,7 +755,7 @@ mod tests {
             "table.name" => "orders",
         ));
 
-        let conn_props = ConnectorProperties::extract(user_props_mysql).unwrap();
+        let conn_props = ConnectorProperties::extract(user_props_mysql, true).unwrap();
         if let ConnectorProperties::MysqlCdc(c) = conn_props {
             assert_eq!(c.props.get("connector_node_addr").unwrap(), "localhost");
             assert_eq!(c.props.get("database.hostname").unwrap(), "127.0.0.1");
@@ -750,7 +768,7 @@ mod tests {
             panic!("extract cdc config failed");
         }
 
-        let conn_props = ConnectorProperties::extract(user_props_postgres).unwrap();
+        let conn_props = ConnectorProperties::extract(user_props_postgres, true).unwrap();
         if let ConnectorProperties::PostgresCdc(c) = conn_props {
             assert_eq!(c.props.get("connector_node_addr").unwrap(), "localhost");
             assert_eq!(c.props.get("database.hostname").unwrap(), "127.0.0.1");

src/connector/src/source/cdc/mod.rs

@@ -78,7 +78,10 @@ pub struct CdcProperties<T: CdcSourceTypeTrait> {
 }
 
 impl<T: CdcSourceTypeTrait> TryFromHashmap for CdcProperties<T> {
-    fn try_from_hashmap(props: HashMap<String, String>) -> anyhow::Result<Self> {
+    fn try_from_hashmap(
+        props: HashMap<String, String>,
+        _deny_unknown_fields: bool,
+    ) -> anyhow::Result<Self> {
         Ok(CdcProperties {
             props,
             table_schema: Default::default(),
@@ -124,6 +127,13 @@ where
     }
 }
 
+impl<T: CdcSourceTypeTrait> crate::source::UnknownFields for CdcProperties<T> {
+    fn unknown_fields(&self) -> HashMap<String, String> {
+        // FIXME: CDC does not handle unknown fields yet
+        HashMap::new()
+    }
+}
+
 impl<T: CdcSourceTypeTrait> CdcProperties<T> {
     pub fn get_source_type_pb(&self) -> SourceType {
         SourceType::from(T::source_type())

src/connector/src/source/datagen/mod.rs

@@ -65,6 +65,13 @@ impl SourceProperties for DatagenProperties {
     const SOURCE_NAME: &'static str = DATAGEN_CONNECTOR;
 }
 
+impl crate::source::UnknownFields for DatagenProperties {
+    fn unknown_fields(&self) -> HashMap<String, String> {
+        // FIXME: datagen does not handle unknown fields yet
+        HashMap::new()
+    }
+}
+
 fn default_rows_per_second() -> u64 {
     10
 }

src/connector/src/source/filesystem/s3/enumerator.rs

@@ -136,6 +136,7 @@ mod tests {
             access: None,
             secret: None,
             endpoint_url: None,
+            unknown_fields: Default::default(),
         };
         let mut enumerator =
             S3SplitEnumerator::new(props.clone(), SourceEnumeratorContext::default().into())

src/connector/src/source/filesystem/s3/mod.rs

@@ -13,19 +13,20 @@
 // limitations under the License.
 mod enumerator;
 
+use std::collections::HashMap;
+
 pub use enumerator::S3SplitEnumerator;
 mod source;
 use serde::Deserialize;
 pub use source::S3FileReader;
 
 use crate::common::AwsAuthProps;
 use crate::source::filesystem::FsSplit;
-use crate::source::SourceProperties;
+use crate::source::{SourceProperties, UnknownFields};
 
 pub const S3_CONNECTOR: &str = "s3";
 
 #[derive(Clone, Debug, Deserialize)]
-#[serde(deny_unknown_fields)]
 pub struct S3Properties {
     #[serde(rename = "s3.region_name")]
     pub region_name: String,
@@ -39,6 +40,9 @@ pub struct S3Properties {
     pub secret: Option<String>,
     #[serde(rename = "s3.endpoint_url")]
     pub endpoint_url: Option<String>,
+
+    #[serde(flatten)]
+    pub unknown_fields: HashMap<String, String>,
 }
 
 impl SourceProperties for S3Properties {
@@ -49,6 +53,12 @@ impl SourceProperties for S3Properties {
     const SOURCE_NAME: &'static str = S3_CONNECTOR;
 }
 
+impl UnknownFields for S3Properties {
+    fn unknown_fields(&self) -> HashMap<String, String> {
+        self.unknown_fields.clone()
+    }
+}
+
 impl From<&S3Properties> for AwsAuthProps {
     fn from(props: &S3Properties) -> Self {
         Self {

src/connector/src/source/filesystem/s3/source/reader.rs

@@ -264,6 +264,7 @@ mod tests {
             access: None,
             secret: None,
             endpoint_url: None,
+            unknown_fields: Default::default(),
         };
         let mut enumerator =
             S3SplitEnumerator::new(props.clone(), SourceEnumeratorContext::default().into())

src/connector/src/source/google_pubsub/mod.rs

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+use serde::Deserialize;
 
 pub mod enumerator;
 pub mod source;
@@ -29,8 +31,7 @@ use crate::source::SourceProperties;
 pub const GOOGLE_PUBSUB_CONNECTOR: &str = "google_pubsub";
 
 #[serde_as]
-#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Hash, WithOptions)]
-#[serde(deny_unknown_fields)]
+#[derive(Clone, Debug, Deserialize, WithOptions)]
 pub struct PubsubProperties {
     #[serde_as(as = "DisplayFromStr")]
     #[serde(rename = "pubsub.split_count")]
@@ -72,6 +73,9 @@ pub struct PubsubProperties {
     /// more details.
     #[serde(rename = "pubsub.start_snapshot")]
     pub start_snapshot: Option<String>,
+
+    #[serde(flatten)]
+    pub unknown_fields: HashMap<String, String>,
 }
 
 impl SourceProperties for PubsubProperties {
@@ -82,6 +86,12 @@ impl SourceProperties for PubsubProperties {
     const SOURCE_NAME: &'static str = GOOGLE_PUBSUB_CONNECTOR;
 }
 
+impl crate::source::UnknownFields for PubsubProperties {
+    fn unknown_fields(&self) -> HashMap<String, String> {
+        self.unknown_fields.clone()
+    }
+}
+
 impl PubsubProperties {
     /// `initialize_env` sets environment variables read by the `google-cloud-pubsub` crate
     pub(crate) fn initialize_env(&self) {
@@ -121,6 +131,8 @@ mod tests {
             start_offset: None,
             start_snapshot: None,
             subscription: String::from("test-subscription"),
+
+            unknown_fields: Default::default(),
         };
 
         let properties = PubsubProperties {

src/connector/src/source/kafka/mod.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+
 use serde::Deserialize;
 use serde_with::{serde_as, DisplayFromStr};
 
@@ -88,7 +90,6 @@ pub struct RdKafkaPropertiesConsumer {
 }
 
 #[derive(Clone, Debug, Deserialize, WithOptions)]
-#[serde(deny_unknown_fields)]
 pub struct KafkaProperties {
     /// This parameter is not intended to be exposed to users.
     /// This parameter specifies only for one parallelism. The parallelism of kafka source
@@ -127,6 +128,9 @@ pub struct KafkaProperties {
 
     #[serde(flatten)]
     pub rdkafka_properties_consumer: RdKafkaPropertiesConsumer,
+
+    #[serde(flatten)]
+    pub unknown_fields: HashMap<String, String>,
 }
 
 impl SourceProperties for KafkaProperties {
@@ -137,6 +141,12 @@ impl SourceProperties for KafkaProperties {
     const SOURCE_NAME: &'static str = KAFKA_CONNECTOR;
 }
 
+impl crate::source::UnknownFields for KafkaProperties {
+    fn unknown_fields(&self) -> HashMap<String, String> {
+        self.unknown_fields.clone()
+    }
+}
+
 impl KafkaProperties {
     pub fn set_client(&self, c: &mut rdkafka::ClientConfig) {
         self.rdkafka_properties_common.set_client(c);

src/connector/src/source/kinesis/mod.rs

@@ -16,6 +16,8 @@ pub mod enumerator;
 pub mod source;
 pub mod split;
 
+use std::collections::HashMap;
+
 use serde::Deserialize;
 use with_options::WithOptions;
 
@@ -28,7 +30,6 @@ use crate::source::SourceProperties;
 pub const KINESIS_CONNECTOR: &str = "kinesis";
 
 #[derive(Clone, Debug, Deserialize, WithOptions)]
-#[serde(deny_unknown_fields)]
 pub struct KinesisProperties {
     #[serde(rename = "scan.startup.mode", alias = "kinesis.scan.startup.mode")]
     // accepted values: "latest", "earliest", "timestamp"
@@ -39,6 +40,9 @@ pub struct KinesisProperties {
 
     #[serde(flatten)]
     pub common: KinesisCommon,
+
+    #[serde(flatten)]
+    pub unknown_fields: HashMap<String, String>,
 }
 
 impl SourceProperties for KinesisProperties {
@@ -48,3 +52,9 @@ impl SourceProperties for KinesisProperties {
 
     const SOURCE_NAME: &'static str = KINESIS_CONNECTOR;
 }
+
+impl crate::source::UnknownFields for KinesisProperties {
+    fn unknown_fields(&self) -> HashMap<String, String> {
+        self.unknown_fields.clone()
+    }
+}

src/connector/src/source/kinesis/source/reader.rs

@@ -322,6 +322,8 @@ mod tests {
 
             scan_startup_mode: None,
             timestamp_offset: Some(123456789098765432),
+
+            unknown_fields: Default::default(),
         };
         let client = KinesisSplitReader::new(
             properties,
@@ -355,6 +357,8 @@ mod tests {
 
             scan_startup_mode: None,
             timestamp_offset: None,
+
+            unknown_fields: Default::default(),
         };
 
         let trim_horizen_reader = KinesisSplitReader::new(