fix: handle kafka sink message timeout error (#12350)

risingwavelabs · Sep 17, 2023 · a975d93 · a975d93
1 parent 8ef74ad
commit a975d93
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/src/connector/src/sink/kafka.rs b/src/connector/src/sink/kafka.rs
@@ -434,7 +434,14 @@ impl KafkaSinkWriter {
                 Err((e, rec)) => {
                     record = rec;
                     match e {
-                        KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull) => {
+                        err @ KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull)
+                        | err @ KafkaError::MessageProduction(RDKafkaErrorCode::MessageTimedOut) => {
+                            tracing::warn!(
+                                "producing message (key {:?}) to topic {} failed, err {:?}, retrying",
+                                record.key.map(|k| k.to_bytes()),
+                                record.topic,
+                                err
+                            );
                             tokio::time::sleep(self.config.retry_interval).await;
                             continue;
                         }