use ArcStr in Kafka offsets (#7747)

GitOrigin-RevId: 54ca3575900bc5ee346d791ba3d49ad6a28daf79

src/connectors/data_storage.rs

@@ -30,6 +30,7 @@ use std::sync::Arc;
 use std::thread::sleep;
 use std::time::{Duration, Instant, SystemTime};
 
+use arcstr::ArcStr;
 use chrono::DateTime;
 use futures::StreamExt;
 use itertools::Itertools;
@@ -767,7 +768,7 @@ impl Writer for FileWriter {
 pub struct KafkaReader {
     consumer: BaseConsumer<DefaultConsumerContext>,
     persistent_id: Option<PersistentId>,
-    topic: Arc<String>,
+    topic: ArcStr,
     positions_for_seek: HashMap<i32, KafkaOffset>,
 }
 
@@ -830,7 +831,7 @@ impl Reader for KafkaReader {
                 if self.topic != *topic {
                     warn!(
                         "Unexpected topic name. Expected: {}, Got: {topic}",
-                        *self.topic
+                        self.topic
                     );
                     continue;
                 }
@@ -880,7 +881,7 @@ impl KafkaReader {
         KafkaReader {
             consumer,
             persistent_id,
-            topic: Arc::new(topic),
+            topic: topic.into(),
             positions_for_seek,
         }
     }

src/connectors/offset.rs

@@ -4,6 +4,7 @@ use std::os::unix::ffi::OsStrExt;
 use std::path::PathBuf;
 use std::sync::Arc;
 
+use arcstr::ArcStr;
 use serde::{Deserialize, Serialize};
 use xxhash_rust::xxh3::Xxh3 as Hasher;
 
@@ -12,7 +13,7 @@ use crate::engine::value::HashInto;
 #[allow(clippy::module_name_repetitions)]
 #[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize, Deserialize, Ord, PartialOrd)]
 pub enum OffsetKey {
-    Kafka(Arc<String>, i32),
+    Kafka(ArcStr, i32),
     Nats(usize),
     Empty,
 }