fix: do not apply empty snapshots

Author: tomverelst

README.md

@@ -101,7 +101,7 @@ All steps can be chained together for streamlined execution:
 
 ```bash
 kbridge fetch -b localhost:9092 | \
-kbridge calculate -b localhost:9093 -l Offset | \
+kbridge calculate -b localhost:9093 -H Offset | \
 kbridge apply -b localhost:9093
 ```
 

bridge-cli/src/main.rs

@@ -84,23 +84,35 @@ async fn run(args: Args) -> Result<(), BridgeError> {
         } => {
             let topics = kafka_connection.topics.clone();
             let client: KafkaBridgeClient = kafka_connection.into();
-            trace!("applying target offsets");
-            let snapshot = OffsetSnapshot::from_csv(input)?;
 
-            // Handle confirmation in CLI before calling the core library
-            if !dry_run && !skip_confirmation && !ask_for_confirmation(&snapshot) {
-                return Err(BridgeError::Message(
-                    "Operation cancelled by user".to_string(),
+            let snapshot = {
+                let snapshot = OffsetSnapshot::from_csv(input)?;
+                snapshot.filter_by_topics(&topics)
+            };
+
+            if snapshot.is_empty() {
+                // TODO decouple from command error
+                return Err(BridgeError::ApplyOffsets(
+                    bridge_core::commands::apply_target_offsets::errors::ApplyOffsetsError::NoOffsetsToApply,
                 ));
             }
 
-            if dry_run {
-                println!("DRY RUN: Would apply the following offsets:");
-                print_offset_snapshot(&snapshot);
-                Ok(())
-            } else {
-                client.apply_target_offsets(topics, snapshot, false).await
+            // Only show confirmation/dry-run output if snapshot has data
+            // (core library validates if offsets remain after topic filtering)
+            if !snapshot.is_empty() {
+                if !dry_run && !skip_confirmation && !ask_for_confirmation(&snapshot) {
+                    return Err(BridgeError::Message(
+                        "Operation cancelled by user".to_string(),
+                    ));
+                }
+
+                if dry_run {
+                    eprintln!("DRY RUN: Would apply the following offsets:");
+                    print_offset_snapshot(&snapshot);
+                }
             }
+
+            client.apply_target_offsets(topics, snapshot, dry_run).await
         }
     }
 }

bridge-core/src/commands/apply_target_offsets/errors.rs

@@ -8,4 +8,6 @@ pub enum ApplyOffsetsError {
     KafkaError(#[from] KafkaError),
     #[error("IO error occurred. Reason: {0}")]
     IoError(#[from] io::Error),
+    #[error("No offsets to apply (input is empty or no matching topics)")]
+    NoOffsetsToApply,
 }

bridge-core/src/commands/apply_target_offsets/mod.rs

@@ -40,6 +40,10 @@ pub async fn execute(
                 .or_insert(vec![value.clone()]);
         });
 
+    if mapped_intermediary_result.is_empty() {
+        return Err(ApplyOffsetsError::NoOffsetsToApply);
+    }
+
     if !dry_run {
         apply_target_offsets(&mut exporter_base_config, &mapped_intermediary_result).await?;
     }

bridge-core/src/lib.rs

@@ -1,5 +1,5 @@
 pub mod client;
-mod commands;
+pub mod commands;
 pub mod errors;
 pub mod kafka;
 pub mod partition;