Repository: risingwavelabs/risingwave (language: rust)

Issue:
Bug: false alarm on nats ack message
### Describe the bug

reported from community: https://risingwave-community.slack.com/archives/C05QRGMATT3/p1737115537124689

> I successfully pull data from nats with the default nats source configuration but if I restart my pod I am loosing data. So I went ahead and recreated the source with consumer.ack_policy='explicit'.
> Data is still flowing in but now risingwave generates some weird error logging.

```
2025-01-17T11:59:05.739773607Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350863.13497.1737115145506842069.0"
2025-01-17T11:59:05.740396932Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350862.13496.1737115145483841558.0"
2025-01-17T11:59:05.74056856Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350864.13498.1737115145509157273.0"
2025-01-17T11:59:05.740630293Z  INFO            rw-streaming async_nats: event: closed
2025-01-17T11:59:05.741280628Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350865.13499.1737115145583432939.0"
2025-01-17T11:59:05.742158035Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350866.13500.1737115145590867515.0"
2025-01-17T11:59:05.742210744Z  INFO            rw-streaming async_nats: event: closed
2025-01-17T11:59:06.697533284Z  INFO            rw-streaming async_nats: event: connected
2025-01-17T11:59:06.699409605Z  INFO            rw-streaming async_nats: event: connected
2025-01-17T11:59:06.748654221Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350869.13503.1737115145752572472.0"
2025-01-17T11:59:06.748711542Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350867.13501.1737115145700876117.0"
2025-01-17T11:59:06.749695035Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350868.13502.1737115145709595317.0"
2025-01-17T11:59:06.749756618Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350870.13504.1737115145754862836.0"
2025-01-17T11:59:06.75079944Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350873.13507.1737115145842816046.0"
2025-01-17T11:59:06.750842473Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350871.13505.1737115145758913510.0"
2025-01-17T11:59:06.751963139Z ERROR            rw-streaming risingwave_connector::source: failed to ack NATS JetStream message error=publish failed: EOF while parsing a value at line 1 column 0 subject="$JS.ACK.SENSOR_DATA.rw2.1.7350872.13506.1
```

but the nats stream does not have data loss

### Error message/log

```text

```

### To Reproduce

_No response_

### Expected behavior

_No response_

### How did you deploy RisingWave?

_No response_

### The version of RisingWave

_No response_

### Additional context

_No response_


Discussion / hints:

Fix patch:
diff --git a/src/connector/src/source/mod.rs b/src/connector/src/source/mod.rs
index 3ea03f7b4df19..af39c598b6c76 100644
--- a/src/connector/src/source/mod.rs
+++ b/src/connector/src/source/mod.rs
@@ -166,14 +166,12 @@ impl WaitCheckpointTask {
                 ref ack_policy,
             ) => {
                 async fn ack(context: &JetStreamContext, reply_subject: String) {
-                    match context.publish(reply_subject.clone(), "".into()).await {
+                    match context.publish(reply_subject.clone(), "+ACK".into()).await {
                         Err(e) => {
                             tracing::error!(error = %e.as_report(), subject = ?reply_subject, "failed to ack NATS JetStream message");
                         }
                         Ok(ack_future) => {
-                            if let Err(e) = ack_future.into_future().await {
-                                tracing::error!(error = %e.as_report(), subject = ?reply_subject, "failed to ack NATS JetStream message");
-                            }
+                            let _ = ack_future.into_future().await;
                         }
                     }
                 }
@@ -193,6 +191,9 @@ impl WaitCheckpointTask {
                     JetStreamAckPolicy::None => (),
                     JetStreamAckPolicy::Explicit => {
                         for reply_subject in reply_subjects {
+                            if reply_subject.is_empty() {
+                                continue;
+                            }
                             ack(context, reply_subject).await;
                         }
                     }


Test patch:
diff --git a/e2e_test/source_inline/nats/operation.py b/e2e_test/source_inline/nats/operation.py
index b33a6fc1d2c15..e34a8f7190b9b 100644
--- a/e2e_test/source_inline/nats/operation.py
+++ b/e2e_test/source_inline/nats/operation.py
@@ -72,6 +72,7 @@ def validate_state_table_item(table_name: str, expect_count: int):
         database="dev"
     )
     # query for the internal table name and make sure it exists
+    result_count = 0
     with conn.cursor() as cursor:
         cursor.execute(f"select name from rw_internal_table_info where job_name = '{table_name}'")
         results = cursor.fetchall()
@@ -82,11 +83,31 @@ def validate_state_table_item(table_name: str, expect_count: int):
             cursor.execute(f"SELECT * FROM {internal_table_name}")
             results = cursor.fetchall()
             print(f"Get items from state table: {results}")
-            if len(results) == expect_count:
+            result_count = len(results)
+            if result_count == expect_count:
                 print(f"Found {expect_count} items in {internal_table_name}")
-                break
-            print(f"Waiting for {internal_table_name} to have {expect_count} items, got {len(results)}. Retry...")
+                return
+            print(f"Waiting for {internal_table_name} to have {expect_count} items, got {result_count}. Retry...")
             time.sleep(0.5)
+    raise Exception(f"Failed to find {expect_count} items in {internal_table_name}, expected {expect_count}, got {result_count}")
+
+
+async def ensure_all_ack(stream_name: str, consumer_name: str):
+    nc = NATS()
+    await nc.connect(servers=[NATS_SERVER])
+    js = nc.jetstream()
+
+    # Get consumer info
+    consumer = await js.consumer_info(stream_name, consumer_name)
+    print(f"Consumer stats for {consumer_name} on stream {stream_name}:")
+    print(f"  Delivered: {consumer.delivered.consumer_seq}")
+    print(f"  Ack Pending: {consumer.num_pending}")
+    print(f"  Redelivered: {consumer.num_redelivered}")
+    print(f"  Waiting: {consumer.num_waiting}")
+
+    if consumer.num_pending > 0:
+        raise Exception(f"Consumer {consumer_name} on stream {stream_name} has {consumer.num_pending} pending messages")
+    await nc.close()
 
 if __name__ == "__main__":
     import sys
@@ -113,7 +134,14 @@ def validate_state_table_item(table_name: str, expect_count: int):
         table_name = sys.argv[2]
         expect_count = int(sys.argv[3])
         validate_state_table_item(table_name, expect_count)
+    elif command == "ensure_all_ack":
+        if len(sys.argv) != 4:
+            print("Error: Both stream name and consumer name are required")
+            sys.exit(1)
+        stream_name = sys.argv[2]
+        consumer_name = sys.argv[3]
+        asyncio.run(ensure_all_ack(stream_name, consumer_name))
     else:
         print(f"Unknown command: {command}")
-        print("Supported commands: create_stream, produce_stream, validate_state")
+        print("Supported commands: create_stream, produce_stream, validate_state, ensure_all_ack")
         sys.exit(1)
diff --git a/e2e_test/source_inline/nats/test.slt.serial b/e2e_test/source_inline/nats/test.slt.serial
index 51519bbcb988b..c989b01725352 100644
--- a/e2e_test/source_inline/nats/test.slt.serial
+++ b/e2e_test/source_inline/nats/test.slt.serial
@@ -39,16 +39,19 @@ select count(*) >= 100 from t_nats;
 ----
 t
 
-system ok
+system ok retry 3 backoff 5s
 python3 e2e_test/source_inline/nats/operation.py validate_state "t_nats" 4
 
+system ok
+python3 e2e_test/source_inline/nats/operation.py ensure_all_ack "teststream" "demo1"
+
 statement ok
 alter table t_nats set PARALLELISM to 6;
 
 system ok
 python3 e2e_test/source_inline/nats/operation.py produce_stream "teststream" "testsubject"
 
-sleep 3s
+sleep 4s
 
 statement ok
 flush;
@@ -58,16 +61,19 @@ select count(*) >= 200 from t_nats;
 ----
 t
 
-system ok
+system ok retry 3 backoff 5s
 python3 e2e_test/source_inline/nats/operation.py validate_state "t_nats" 6
 
+system ok
+python3 e2e_test/source_inline/nats/operation.py ensure_all_ack "teststream" "demo1"
+
 statement ok
 alter table t_nats set PARALLELISM to 2;
 
 system ok
 python3 e2e_test/source_inline/nats/operation.py produce_stream "teststream" "testsubject"
 
-sleep 3s
+sleep 4s
 
 statement ok
 flush;
@@ -77,9 +83,12 @@ select count(*) >= 300 from t_nats;
 ----
 t
 
-system ok
+system ok retry 3 backoff 5s
 python3 e2e_test/source_inline/nats/operation.py validate_state "t_nats" 2
 
+system ok
+python3 e2e_test/source_inline/nats/operation.py ensure_all_ack "teststream" "demo1"
+
 statement ok
 drop table t_nats;