feat: MM2 features, string/cache transforms, hardened error handling

## New features

### MirrorMaker 2 parity
- Topic regex subscription: `input: "^payments.*"` subscribes to all matching
  topics and picks up newly created ones automatically (rdkafka native support)
- Output topic templates: `output: "mirror.{source_topic}"` resolves per-message
  from the envelope source topic with lazy partition-count caching

### Cache transforms (CACHE_PUT / CACHE_LOOKUP)
- `SyncLookupCache` / `SyncCacheManager` — sync Moka cache shared across the
  transform pipeline; named stores auto-created on first reference
- `CACHE_PUT:/key,store[,/valuePath]` — write message (or a field) to a named
  store; message passes through unchanged
- `CACHE_LOOKUP:/key,store,field|MERGE` — enrich message from cache; pass
  through on miss or missing key

### String transforms (STRING:*)
- UPPER, LOWER, TRIM, TRIM_START, TRIM_END, LENGTH, SUBSTRING, REPLACE,
  REPLACE_ALL, REGEX_REPLACE, SPLIT, CONCAT (12 operations)
- All accept an optional `,outputField` to write result without touching original

### Single-destination transform support
- New `transform: <expr>` top-level config field for single-destination mode
- Full DSL (including CACHE_*) available without switching to routing mode

## Error handling hardening

### Fail-soft on data — all transforms pass through on missing/null fields
- Missing field, null value, or wrong type for the operation → message passes
  through unchanged with a debug log; pipeline never halted for data issues
- Applies to: JsonPath, ArrayMap, Arithmetic, Hash, String, Concat, CacheLookup,
  CachePut transforms
- Config errors (invalid regex, empty `from`, empty CONCAT parts) still fail
  at parse time

### Additional fixes
- `resolve_topic`: template + no source topic → error (was silent `mirror.unknown`)
- `get_or_fetch_partitions`: blocking `fetch_metadata` wrapped in `block_in_place`
- `partition_cache` mutex uses `unwrap_or_else(|e| e.into_inner())` for poison recovery
- `flush()`: errors now propagate via `MirrorMakerError::Kafka` (was `let _ = ...`)
- `flush_all`: attempts all sinks, collects errors, never abandons remaining sinks
- `parse_input_topics`: warns on apparent regex patterns without `^` prefix
- `STRING:REPLACE`: empty `from` rejected at parse time (prevents per-char insertion)
- `STRING:CONCAT`: empty parts rejected at parse time (trailing/double commas)
- `STRING:CONCAT`: empty `outputField` rejected at parse time
- `routing` + `transform` both set: `warn!` that top-level transform is ignored

## README
- Comprehensive comparison vs MirrorMaker 1 and MirrorMaker 2 (feature matrix,
  performance table, when-to-choose guidance)

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

Author: rahulbsw

Cargo.toml

@@ -53,7 +53,7 @@ digest = "0.10"
 hex = "0.4"
 
 # Caching
-moka = { version = "0.12", features = ["future"] }
+moka = { version = "0.12", features = ["future", "sync"] }
 dashmap = "6.0"  # Concurrent HashMap
 redis = { version = "0.24", features = ["tokio-comp", "connection-manager"], optional = true }
 

README.md

@@ -186,6 +186,112 @@ impl Default for CacheManager {
     }
 }
 
+// ============================================================================
+// Sync cache — used by transform/filter pipeline (which is synchronous)
+// ============================================================================
+
+/// Synchronous lookup cache backed by `moka::sync::Cache`.
+///
+/// Used by `CacheLookupTransform` and `CachePutTransform` so they can
+/// implement the synchronous `Transform` trait without blocking an async thread.
+pub struct SyncLookupCache {
+    cache: moka::sync::Cache<String, Value>,
+}
+
+impl SyncLookupCache {
+    pub fn new(config: CacheConfig) -> Self {
+        let mut builder = moka::sync::Cache::builder()
+            .max_capacity(config.max_capacity);
+
+        if let Some(ttl) = config.ttl_seconds {
+            builder = builder.time_to_live(Duration::from_secs(ttl));
+        }
+        if let Some(tti) = config.tti_seconds {
+            builder = builder.time_to_idle(Duration::from_secs(tti));
+        }
+
+        info!(
+            "Created sync lookup cache: max_capacity={}, ttl={:?}s, tti={:?}s",
+            config.max_capacity, config.ttl_seconds, config.tti_seconds
+        );
+
+        Self { cache: builder.build() }
+    }
+
+    /// Look up a value by key. Returns `None` on cache miss.
+    pub fn get(&self, key: &str) -> Option<Value> {
+        let result = self.cache.get(key);
+        if result.is_some() {
+            debug!("Cache hit: {}", key);
+        } else {
+            debug!("Cache miss: {}", key);
+        }
+        result
+    }
+
+    /// Insert a value into the cache.
+    pub fn put(&self, key: String, value: Value) {
+        debug!("Cache put: {}", key);
+        self.cache.insert(key, value);
+    }
+
+    /// Remove a key from the cache.
+    pub fn remove(&self, key: &str) {
+        self.cache.invalidate(key);
+    }
+
+    /// Returns true if the key is present.
+    pub fn contains_key(&self, key: &str) -> bool {
+        self.cache.contains_key(key)
+    }
+
+    pub fn entry_count(&self) -> u64 {
+        self.cache.entry_count()
+    }
+}
+
+/// Named store of `SyncLookupCache` instances.
+///
+/// Passed to the transform parser so `CACHE_LOOKUP` and `CACHE_PUT`
+/// expressions can reference caches by name. Caches are created on
+/// first use with default settings (10k entries, 1h TTL).
+pub struct SyncCacheManager {
+    caches: dashmap::DashMap<String, Arc<SyncLookupCache>>,
+}
+
+impl SyncCacheManager {
+    pub fn new() -> Self {
+        Self { caches: dashmap::DashMap::new() }
+    }
+
+    /// Get an existing cache by name, or create it with default config.
+    pub fn get_or_create(&self, name: &str) -> Arc<SyncLookupCache> {
+        self.caches
+            .entry(name.to_string())
+            .or_insert_with(|| {
+                info!("Creating new sync cache store: '{}'", name);
+                Arc::new(SyncLookupCache::new(CacheConfig::default()))
+            })
+            .clone()
+    }
+
+    /// Get an existing cache by name. Returns `None` if it doesn't exist.
+    pub fn get(&self, name: &str) -> Option<Arc<SyncLookupCache>> {
+        self.caches.get(name).map(|e| e.clone())
+    }
+
+    /// List all named stores.
+    pub fn store_names(&self) -> Vec<String> {
+        self.caches.iter().map(|e| e.key().clone()).collect()
+    }
+}
+
+impl Default for SyncCacheManager {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/cache.rs

@@ -34,6 +34,14 @@ pub struct MirrorMakerConfig {
     /// Multi-destination routing configuration
     pub routing: Option<RoutingConfig>,
 
+    /// Value transform expression for single-destination mode.
+    ///
+    /// Supports the full transform DSL including `STRING:`, `CACHE_LOOKUP:`,
+    /// `CACHE_PUT:`, `HASH:`, `CONSTRUCT:`, `ARITHMETIC:`, etc.
+    /// Ignored when `routing` is set.
+    #[serde(default)]
+    pub transform: Option<String>,
+
     /// Consumer properties
     #[serde(default)]
     pub consumer_properties: HashMap<String, String>,