Add tests for the kafka_producer, ensuring error behaviour on startup
use super::super::ConnectorHarness;
use super::redpanda_container;
+use crate::connectors::tests::free_port::find_free_tcp_port;
use crate::{connectors::impls::kafka, errors::Result, Event};
use async_std::prelude::FutureExt;
use futures::StreamExt;
@@ -272,3 +273,63 @@ async fn connector_kafka_producer() -> Result<()> {
drop(container);
Ok(())
}
+
+#[async_std::test]
+#[serial(kafka, timeout_ms = 600000)]
+async fn producer_unreachable() -> Result<()> {
+ let _ = env_logger::try_init();
+ let port = find_free_tcp_port().await?;
+ let broker = format!("127.0.0.1:{port}");
+ let topic = "unreachable";
+ let connector_config = literal!({
+ "codec": "json-sorted",
+ "config": {
+ "brokers": [
+ broker.clone()
+ ],
+ "topic": topic,
+ "key": "snot",
+ "rdkafka_options": {
+ "debug": "all"
+ }
+ }
+ });
+ let harness = ConnectorHarness::new(
+ function_name!(),
+ &kafka::producer::Builder::default(),
+ &connector_config,
+ )
+ .await?;
+ assert!(harness.start().await.is_err());
+ Ok(())
+}
+
+#[async_std::test]
+#[serial(kafka, timeout_ms = 600000)]
+async fn producer_unresolvable() -> Result<()> {
+ let _ = env_logger::try_init();
+ let port = find_free_tcp_port().await?;
+ let broker = format!("i_do_not_resolve:{port}");
+ let topic = "unresolvable";
+ let connector_config = literal!({
+ "codec": "json-sorted",
+ "config": {
+ "brokers": [
+ broker.clone()
+ ],
+ "topic": topic,
+ "key": "snot",
+ "rdkafka_options": {
+ "debug": "all"
+ }
+ }
+ });
+ let harness = ConnectorHarness::new(
+ function_name!(),
+ &kafka::producer::Builder::default(),
+ &connector_config,
+ )
+ .await?;
+ assert!(harness.start().await.is_err());
+ Ok(())
+}
