chore: Improve wording of error message

e2e_test/source/basic/ddl.slt

@@ -53,7 +53,7 @@ create source s (
   properties.bootstrap.server = 'message_queue:29092'
 ) FORMAT PLAIN ENCODE JSON;
 
-statement error properties `scan_startup_mode` only support earliest and latest or leave it empty
+statement error properties `scan_startup_mode` only supports earliest and latest or leaving it empty
 create source invalid_startup_mode (
   column1 varchar
 ) with (

e2e_test/source/basic/old_row_format_syntax/ddl.slt

@@ -7,7 +7,7 @@ create source s (
   properties.bootstrap.server = 'message_queue:29092'
 ) ROW FORMAT JSON;
 
-statement error properties `scan_startup_mode` only support earliest and latest or leave it empty
+statement error properties `scan_startup_mode` only supports earliest and latest or leaving it empty
 create source invalid_startup_mode (
   column1 varchar
 ) with (

java/connector-node/risingwave-sink-cassandra/src/main/java/com/risingwave/connector/CassandraUtil.java

@@ -91,7 +91,7 @@ public static void checkSchema(
                 throw Status.FAILED_PRECONDITION
                         .withDescription(
                                 String.format(
-                                        "Don't match in the name, rw is %s cassandra can't find it",
+                                        "Name mismatch. %s of RisingWave is not found in Cassandra.",
                                         columnDesc.getName()))
                         .asRuntimeException();
             }

java/connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-common/src/main/java/com/risingwave/mock/flink/common/FlinkDynamicUtil.java

@@ -120,14 +120,14 @@ public static void validateSchemaWithCatalog(
                 throw Status.FAILED_PRECONDITION
                         .withDescription(
                                 String.format(
-                                        "Don't match in the name, rw is %s", columnDesc.getName()))
+                                        "Name mismatch. RisingWave is %s", columnDesc.getName()))
                         .asRuntimeException();
             }
             if (!checkType(columnDesc.getDataType(), flinkColumnMap.get(columnDesc.getName()))) {
                 throw Status.FAILED_PRECONDITION
                         .withDescription(
                                 String.format(
-                                        "Don't match in the type, name is %s, Sink is %s, rw is %s",
+                                        "Type mismatch. Name is %s, Sink is %s, RisingWave is %s",
                                         columnDesc.getName(),
                                         flinkColumnMap.get(columnDesc.getName()),
                                         columnDesc.getDataType().getTypeName()))

src/connector/src/sink/log_store.rs

@@ -85,7 +85,7 @@ impl TruncateOffset {
             } => {
                 if epoch != *offset_epoch {
                     bail!(
-                        "new item epoch {} not match current chunk offset epoch {}",
+                        "new item epoch {} does not match current chunk offset epoch {}",
                         epoch,
                         offset_epoch
                     );

src/connector/src/sink/mqtt.rs

@@ -133,7 +133,7 @@ impl MqttConfig {
             .map_err(|e| SinkError::Config(anyhow!(e)))?;
         if config.r#type != SINK_TYPE_APPEND_ONLY {
             Err(SinkError::Config(anyhow!(
-                "Mqtt sink only support append-only mode"
+                "Mqtt sink only supports append-only mode"
             )))
         } else {
             Ok(config)
@@ -175,7 +175,7 @@ impl Sink for MqttSink {
     async fn validate(&self) -> Result<()> {
         if !self.is_append_only {
             return Err(SinkError::Mqtt(anyhow!(
-                "Mqtt sink only support append-only mode"
+                "Mqtt sink only supports append-only mode"
             )));
         }
 
@@ -261,7 +261,7 @@ impl MqttSinkWriter {
             },
             _ => {
                 return Err(SinkError::Config(anyhow!(
-                    "Mqtt sink only support append-only mode"
+                    "Mqtt sink only supports append-only mode"
                 )))
             }
         };

src/connector/src/sink/nats.rs

@@ -79,7 +79,7 @@ impl NatsConfig {
             .map_err(|e| SinkError::Config(anyhow!(e)))?;
         if config.r#type != SINK_TYPE_APPEND_ONLY {
             Err(SinkError::Config(anyhow!(
-                "Nats sink only support append-only mode"
+                "Nats sink only supports append-only mode"
             )))
         } else {
             Ok(config)

src/connector/src/source/kafka/enumerator/client.rs

@@ -82,7 +82,7 @@ impl SplitEnumerator for KafkaSplitEnumerator {
             Some("latest") => KafkaEnumeratorOffset::Latest,
             None => KafkaEnumeratorOffset::Earliest,
             _ => bail!(
-                "properties `scan_startup_mode` only support earliest and latest or leave it empty"
+                "properties `scan_startup_mode` only supports earliest and latest or leaving it empty"
             ),
         };
 

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

@@ -199,7 +199,7 @@ impl KinesisSplitReader {
                 }
                 Err(e) => {
                     let error = anyhow!(e).context(format!(
-                        "Kinesis got a unhandled error on stream {:?}, shard {:?}",
+                        "Kinesis got an unhandled error on stream {:?}, shard {:?}",
                         self.stream_name, self.shard_id
                     ));
                     tracing::error!(error = %error.as_report());

src/connector/src/source/pulsar/enumerator/client.rs

@@ -65,7 +65,7 @@ impl SplitEnumerator for PulsarSplitEnumerator {
             None => PulsarEnumeratorOffset::Earliest,
             _ => {
                 bail!(
-                    "properties `startup_mode` only support earliest and latest or leave it empty"
+                    "properties `startup_mode` only supports earliest and latest or leaving it empty"
                 );
             }
         };

src/meta/src/controller/fragment.rs

@@ -210,7 +210,7 @@ impl CatalogController {
         for mut actor in pb_actors {
             let mut upstream_actors = BTreeMap::new();
 
-            let node = actor.nodes.as_mut().context("nodes is empty")?;
+            let node = actor.nodes.as_mut().context("nodes are empty")?;
 
             visit_stream_node(node, |body| {
                 if let NodeBody::Merge(m) = body {

src/meta/src/manager/catalog/database.rs

@@ -255,7 +255,7 @@ impl DatabaseManager {
                 && x.name.eq(&relation_key.2)
         }) {
             if t.stream_job_status == StreamJobStatus::Creating as i32 {
-                bail!("table is in creating procedure, table id: {}", t.id);
+                bail!("Creating the table, table id: {}", t.id);
             } else {
                 Err(MetaError::catalog_duplicated("table", &relation_key.2))
             }

src/meta/src/manager/catalog/fragment.rs

@@ -320,7 +320,7 @@ impl FragmentManager {
         let map = &mut guard.table_fragments;
         let table_id = table_fragment.table_id();
         if map.contains_key(&table_id) {
-            bail!("table_fragment already exist: id={}", table_id);
+            bail!("table_fragment already exists: id={}", table_id);
         }
 
         let mut table_fragments = BTreeMapTransaction::new(map);

src/meta/src/rpc/ddl_controller.rs

@@ -386,7 +386,7 @@ impl DdlController {
                 .await?
                 .is_empty()
         {
-            bail!("There are background creating jobs, please try again later")
+            bail!("The system is creating jobs in the background, please try again later")
         }
 
         self.stream_manager

src/stream/src/executor/dynamic_filter.rs

@@ -403,8 +403,8 @@ impl<S: StateStore, const USE_WATERMARK_CACHE: bool> DynamicFilterExecutor<S, US
                         let (range, _latest_is_lower, is_insert) = self.get_range(&curr, prev);
 
                         if !is_insert && self.condition_always_relax {
-                            bail!("The optimizer inferred that the right side's change always make the condition more relaxed.\
-                                But the right changes make the conditions stricter.");
+                            bail!("The optimizer incorrectly assumed that changes on the right side always relax the condition.\
+                                But they actually make it stricter.");
                         }
 
                         let range = (Self::to_row_bound(range.0), Self::to_row_bound(range.1));

src/stream/src/executor/source/state_table_handler.rs

@@ -144,7 +144,7 @@ impl<S: StateStore> SourceStateTableHandler<S> {
     ) -> StreamExecutorResult<()> {
         if states.is_empty() {
             // TODO should be a clear Error Code
-            bail!("states require not null");
+            bail!("states should not be null");
         } else {
             for split in states {
                 self.set_complete(split.id(), split.encode_to_json())

src/stream/src/executor/watermark_filter.rs

@@ -333,7 +333,7 @@ impl<S: StateStore> WatermarkFilterExecutor<S> {
                 if row.len() == 1 {
                     Ok::<_, StreamExecutorError>(row[0].to_owned())
                 } else {
-                    bail!("The watermark row should only contains 1 datum");
+                    bail!("The watermark row should only contain 1 datum");
                 }
             }
             _ => Ok(None),