Repository: tikv/tikv (language: rust)

Issue:
#17656 introduces a panic into TiKV cdc module
## Bug Report

#17656 filters out entries before loading old value if they are not in observed range.

However, it introduces a bug, which may cause TiKV instance panic incorrectly.

### What version of TiKV are you using?
v7.5.5
v7.1.6

### What operating system and CPU are you using?

### Steps to reproduce

### What did you expect?

### What did happened?


Discussion / hints:

Fix patch:
diff --git a/components/cdc/src/delegate.rs b/components/cdc/src/delegate.rs
index f5f28270ba6..82a93e67870 100644
--- a/components/cdc/src/delegate.rs
+++ b/components/cdc/src/delegate.rs
@@ -564,6 +564,7 @@ impl Delegate {
         request_id: u64,
         entries: Vec<Option<KvEntry>>,
         filter_loop: bool,
+        observed_range: &ObservedRange,
     ) -> Result<Vec<CdcEvent>> {
         let entries_len = entries.len();
         let mut rows = vec![Vec::with_capacity(entries_len)];
@@ -581,6 +582,9 @@ impl Delegate {
                     lock,
                     old_value,
                 })) => {
+                    if !observed_range.contains_encoded_key(&lock.0) {
+                        continue;
+                    }
                     let l = Lock::parse(&lock.1).unwrap();
                     if decode_lock(lock.0, l, &mut row, &mut _has_value) {
                         continue;
@@ -594,6 +598,9 @@ impl Delegate {
                     write,
                     old_value,
                 })) => {
+                    if !observed_range.contains_encoded_key(&write.0) {
+                        continue;
+                    }
                     if decode_write(write.0, &write.1, &mut row, &mut _has_value, false) {
                         continue;
                     }
diff --git a/components/cdc/src/initializer.rs b/components/cdc/src/initializer.rs
index c49dc89e661..86aabedc72c 100644
--- a/components/cdc/src/initializer.rs
+++ b/components/cdc/src/initializer.rs
@@ -357,8 +357,8 @@ impl<E: KvEngine> Initializer<E> {
                         if self.observed_range.contains_encoded_key(key) {
                             read_old_value(entry.old_value(), &mut stats.old_value)?;
                             total_bytes += entry.size();
-                            entries.push(Some(KvEntry::TxnEntry(entry)));
                         }
+                        entries.push(Some(KvEntry::TxnEntry(entry)));
                     }
                     None => {
                         entries.push(None);
@@ -443,6 +443,7 @@ impl<E: KvEngine> Initializer<E> {
             self.request_id,
             entries,
             self.filter_loop,
+            &self.observed_range,
         )?;
         if done {
             let (cb, fut) = tikv_util::future::paired_future_callback();
@@ -743,14 +744,12 @@ mod tests {
             total_bytes += v.len();
             let ts = TimeStamp::new(i as _);
             must_prewrite_put(&mut engine, k, v, k, ts);
-            if i < 90 {
-                let txn_locks = expected_locks.entry(ts).or_insert_with(|| {
-                    let mut txn_locks = TxnLocks::default();
-                    txn_locks.sample_lock = Some(k.to_vec().into());
-                    txn_locks
-                });
-                txn_locks.lock_count += 1;
-            }
+            let txn_locks = expected_locks.entry(ts).or_insert_with(|| {
+                let mut txn_locks = TxnLocks::default();
+                txn_locks.sample_lock = Some(k.to_vec().into());
+                txn_locks
+            });
+            txn_locks.lock_count += 1;
         }
 
         let region = Region::default();


Test patch:
diff --git a/components/cdc/tests/integrations/test_cdc.rs b/components/cdc/tests/integrations/test_cdc.rs
index 16fb81b0170..f128cce027a 100644
--- a/components/cdc/tests/integrations/test_cdc.rs
+++ b/components/cdc/tests/integrations/test_cdc.rs
@@ -2730,3 +2730,56 @@ fn test_cdc_filter_key_range() {
 
     suite.stop();
 }
+
+#[test]
+fn test_partial_subscription_build_resolver() {
+    let cluster = new_server_cluster(0, 2);
+    cluster.pd_client.disable_default_operator();
+    let mut suite = TestSuiteBuilder::new().cluster(cluster).build();
+    let rid = suite.cluster.get_region(&[]).id;
+
+    let start_tso = block_on(suite.cluster.pd_client.get_tso()).unwrap();
+    let pk = format!("key_{:05}", 0).into_bytes();
+    let mut mutations = Vec::with_capacity(1000);
+    let mut keys = Vec::with_capacity(1000);
+    for i in 0..1 {
+        let key = format!("key_{:05}", i).into_bytes();
+        keys.push(key.clone());
+
+        let mut mutation = Mutation::default();
+        mutation.set_op(Op::Put);
+        mutation.key = key;
+        mutation.value = vec![b'x'; 16];
+        mutations.push(mutation);
+    }
+    suite.must_kv_prewrite(rid, mutations, pk, start_tso);
+
+    // Subscription a range without any locks.
+    let (mut req_tx, _, receive_event) = new_event_feed(suite.get_region_cdc_client(rid));
+    let mut req = suite.new_changedata_request(rid);
+    req.request_id = 100;
+    req.checkpoint_ts = start_tso.into_inner() - 1;
+    req.set_start_key(Key::from_raw(b"m").into_encoded());
+    req.set_end_key(Key::from_raw(b"z").into_encoded());
+    block_on(req_tx.send((req.clone(), WriteFlags::default()))).unwrap();
+    println!("send request ok");
+
+    // The region subscription can be initialized correctly.
+    let events = receive_event(false).events.to_vec();
+    assert_eq!(events.len(), 1, "{:?}", events);
+    match events[0].event.as_ref().unwrap() {
+        Event_oneof_event::Entries(es) => {
+            assert!(es.entries.len() == 1);
+            assert_eq!(es.entries[0].get_type(), EventLogType::Initialized);
+        }
+        _ => unreachable!(),
+    }
+
+    sleep_ms(1000);
+
+    // However, ResolvedTs shouldn't be advanced incorrectly.
+    let event = receive_event(true);
+    assert!(event.events.is_empty());
+    assert!(event.has_resolved_ts());
+    assert_eq!(event.get_resolved_ts().ts, start_tso.into_inner());
+}