fix: fix kafka wal logs deletion and recovery logic

components/message_queue/src/kafka/kafka_impl.rs

@@ -284,6 +284,7 @@ impl MessageQueue for KafkaImpl {
             })
     }
 
+    // FIXME: consume a empty topic may be hanged forever...
     async fn consume(
         &self,
         topic_name: &str,

components/message_queue/src/tests/cases.rs

@@ -133,8 +133,7 @@ async fn test_consume_empty_topic<T: MessageQueue>(message_queue: &T) {
         .await
         .is_ok());
 
-    // Call produce to push messages at first, then call consume to pull back and
-    // compare.
+    // FIXME: consume a empty topic may be hanged forever...
     let mut iter = message_queue
         .consume(&topic_name, StartOffset::Earliest)
         .await

wal/src/kv_encoder.rs

@@ -1,4 +1,4 @@
-// Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
+// Copyright 2022-2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
 //! Common Encoding for Wal logs
 
@@ -115,17 +115,15 @@ pub enum Namespace {
 }
 
 /// Log key in old wal design, map the `TableId` to `RegionId`
-#[allow(unused)]
+
 pub type LogKey = (u64, SequenceNumber);
 
-#[allow(unused)]
 #[derive(Debug, Clone)]
 pub struct LogKeyEncoder {
     pub version: u8,
     pub namespace: Namespace,
 }
 
-#[allow(unused)]
 impl LogKeyEncoder {
     /// Create newest version encoder.
     pub fn newest() -> Self {
@@ -202,13 +200,11 @@ impl Decoder<LogKey> for LogKeyEncoder {
     }
 }
 
-#[allow(unused)]
 #[derive(Debug, Clone)]
 pub struct LogValueEncoder {
     pub version: u8,
 }
 
-#[allow(unused)]
 impl LogValueEncoder {
     /// Create newest version encoder.
     pub fn newest() -> Self {
@@ -240,12 +236,10 @@ impl<T: Payload> Encoder<T> for LogValueEncoder {
     }
 }
 
-#[allow(unused)]
 pub struct LogValueDecoder {
     pub version: u8,
 }
 
-#[allow(unused)]
 impl LogValueDecoder {
     pub fn decode<'a>(&self, mut buf: &'a [u8]) -> Result<&'a [u8]> {
         let version = buf.try_get_u8().context(DecodeLogValueHeader)?;
@@ -475,7 +469,6 @@ impl MaxSeqMetaEncoding {
     }
 }
 
-#[allow(unused)]
 #[derive(Debug, Clone)]
 pub struct LogEncoding {
     key_enc: LogKeyEncoder,
@@ -484,7 +477,6 @@ pub struct LogEncoding {
     value_enc_version: u8,
 }
 
-#[allow(unused)]
 impl LogEncoding {
     pub fn newest() -> Self {
         Self {
@@ -589,7 +581,7 @@ impl LogBatchEncoder {
 }
 
 /// Common log key used in multiple wal implementation
-#[allow(unused)]
+
 #[derive(Debug, Copy, Clone, Eq, PartialEq, PartialOrd, Ord)]
 pub struct CommonLogKey {
     /// Id of region which the table belongs to,
@@ -599,7 +591,6 @@ pub struct CommonLogKey {
     pub sequence_num: SequenceNumber,
 }
 
-#[allow(unused)]
 impl CommonLogKey {
     pub fn new(region_id: u64, table_id: TableId, sequence_num: SequenceNumber) -> Self {
         Self {
@@ -610,14 +601,12 @@ impl CommonLogKey {
     }
 }
 
-#[allow(unused)]
 #[derive(Debug, Clone)]
 pub struct CommonLogKeyEncoder {
     pub version: u8,
     pub namespace: Namespace,
 }
 
-#[allow(unused)]
 impl CommonLogKeyEncoder {
     /// Create newest version encoder.
     pub fn newest() -> Self {
@@ -697,7 +686,6 @@ impl Decoder<CommonLogKey> for CommonLogKeyEncoder {
     }
 }
 
-#[allow(unused)]
 #[derive(Debug, Clone)]
 pub struct CommonLogEncoding {
     key_enc: CommonLogKeyEncoder,
@@ -706,7 +694,6 @@ pub struct CommonLogEncoding {
     value_enc_version: u8,
 }
 
-#[allow(unused)]
 impl CommonLogEncoding {
     pub fn newest() -> Self {
         Self {

wal/src/message_queue_impl/encoding.rs

@@ -1,4 +1,4 @@
-// Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
+// Copyright 2022-2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
 //! Meta encoding of wal's message queue implementation
 
@@ -90,25 +90,23 @@ pub enum Error {
 define_result!(Error);
 
 /// Generate wal data topic name
-#[allow(unused)]
+
 pub fn format_wal_data_topic_name(namespace: &str, region_id: u64) -> String {
     format!("{namespace}_data_{region_id}")
 }
 
 /// Generate wal meta topic name
-#[allow(unused)]
+
 pub fn format_wal_meta_topic_name(namespace: &str, region_id: u64) -> String {
     format!("{namespace}_meta_{region_id}")
 }
 
-#[allow(unused)]
 #[derive(Clone, Debug)]
 pub struct MetaEncoding {
     key_enc: MetaKeyEncoder,
     value_enc: MetaValueEncoder,
 }
 
-#[allow(unused)]
 impl MetaEncoding {
     pub fn encode_key(&self, buf: &mut BytesMut, meta_key: &MetaKey) -> manager::Result<()> {
         buf.clear();
@@ -153,6 +151,7 @@ impl MetaEncoding {
         Ok(meta_value.into())
     }
 
+    #[allow(dead_code)]
     pub fn is_meta_key(&self, mut buf: &[u8]) -> manager::Result<bool> {
         self.key_enc
             .is_valid(&mut buf)
@@ -174,20 +173,19 @@ impl MetaEncoding {
 }
 
 /// Message queue implementation's meta key
-#[allow(unused)]
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct MetaKey(pub u64);
 
-#[allow(unused)]
 #[derive(Clone, Debug)]
 pub struct MetaKeyEncoder {
     pub namespace: Namespace,
     pub version: u8,
 }
 
-#[allow(unused)]
 impl MetaKeyEncoder {
     /// Determine whether the raw bytes is a valid meta key.
+    #[allow(dead_code)]
     pub fn is_valid<B: Buf>(&self, buf: &mut B) -> Result<bool> {
         let namespace = buf.try_get_u8().context(DecodeMetaKey)?;
         let version = buf.try_get_u8().context(DecodeMetaKey)?;

wal/src/message_queue_impl/namespace.rs

@@ -31,7 +31,7 @@ use crate::{
 #[derive(Debug, Snafu)]
 pub enum Error {
     #[snafu(display(
-        "Failed to open region, namespace:{}, location:{:?}, err:{}",
+        "Failed to get sequence, namespace:{}, location:{:?}, err:{}",
         namespace,
         location,
         source
@@ -43,7 +43,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to open region, namespace:{}, request:{:?}, err:{}",
+        "Failed to read table logs, namespace:{}, request:{:?}, err:{}",
         namespace,
         request,
         source
@@ -56,7 +56,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to open region, namespace:{}, request:{:?}, \nBacktrace:\n{}",
+        "Failed to read table logs, namespace:{}, request:{:?}, \nBacktrace:\n{}",
         namespace,
         request,
         backtrace,
@@ -69,7 +69,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to open region, namespace:{}, request:{:?}, err:{}",
+        "Failed to scan region logs, namespace:{}, request:{:?}, err:{}",
         namespace,
         request,
         source
@@ -82,7 +82,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to open region, namespace:{}, request:{:?}, \nBacktrace:\n{}",
+        "Failed to scan region logs, namespace:{}, request:{:?}, \nBacktrace:\n{}",
         namespace,
         request,
         backtrace,
@@ -95,7 +95,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to open region, namespace:{}, location:{:?}, batch_size:{}, err:{}",
+        "Failed to write logs, namespace:{}, location:{:?}, batch_size:{}, err:{}",
         namespace,
         location,
         batch_size,
@@ -109,7 +109,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to open region, namespace:{}, location:{:?}, sequence_num:{}, err:{}",
+        "Failed to mark logs deleted, namespace:{}, location:{:?}, sequence_num:{}, err:{}",
         namespace,
         location,
         sequence_num,