fix: use match to handle non queue full error instead of assert (#11998)

risingwavelabs · Aug 31, 2023 · e0480ea · e0480ea
1 parent 8ea014f
commit e0480ea
Showing 1 changed file with 7 additions and 4 deletions.
diff --git a/src/connector/src/sink/kafka.rs b/src/connector/src/sink/kafka.rs
@@ -430,10 +430,13 @@ impl KafkaSinkWriter {
                 Err((e, rec)) => {
                     err = e;
                     record = rec;
-                    // FIXME: Will there possibly exist other errors?
-                    assert!(err == KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull));
-                    tokio::time::sleep(self.config.retry_interval).await;
-                    continue;
+                    match err {
+                        KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull) => {
+                            tokio::time::sleep(self.config.retry_interval).await;
+                            continue;
+                        }
+                        _ => break,
+                    }
                 }
             }
         }