2270814456

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/tremor-rs/tremor-runtime/commit/<a class=hub.com/tremor-rs/tremor-runtime/commit/<a class="double-link" href="https://git"><a class=hub.com/tremor-rs/tremor-runtime/commit/<a class="double-link" href="https://git"><a class=hub.com/tremor-rs/tremor-runtime/commit/<a class="double-link" href="https://git"><a class=hub.com/tremor-rs/tremor-runtime/commit/542edd88a2f8dd641d54c20015279d10c7bb257d">542edd88a">&lt;a href=&quot;https://github.com/tremor-rs/tremor-runtime/commit/</a><a class="double-link" href="https://github.com/tremor-rs/tremor-runtime/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tremor-rs/tremor-runtime/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tremor-rs/tremor-runtime/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tremor-rs/tremor-runtime/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/tremor-rs/tremor-runtime/commit/542edd88a2f8dd641d54c20015279d10c7bb257d">542edd88a</a><a href="https://github.com/tremor-rs/tremor-runtime/commit/542edd88a2f8dd641d54c20015279d10c7bb257d">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/tremor-rs/tremor-runtime/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/tremor-rs/tremor-runtime/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/tremor-rs/tremor-runtime/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/tremor-rs/tremor-runtime/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/tremor-rs/tremor-runtime/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/tremor-rs/tremor-runtime/commit/542edd88a2f8dd641d54c20015279d10c7bb257d&quot;&gt;542edd88a&lt;/a&gt;&lt;a href=&quot;https://github.com/tremor-rs/tremor-runtime/commit/542edd88a2f8dd641d54c20015279d10c7bb257d&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/tremor-rs/tremor-runtime/commit/&amp;lt;/a&amp;gt;&amp;lt;a... (continued)

Pull Request Pull Request #1606: Deny unknown connector fields.

Run Details

160 of 160 new or added lines in 13 files covered. (100.0%)

40031 of 46764 relevant lines covered (85.6%)

3721.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.0

/src/connectors/impls/kafka.rs

// Copyright 2022, The Tremor Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
pub(crate) mod consumer;
pub(crate) mod producer;

use crate::errors::{Error, Kind as ErrorKind, Result};
use core::future::Future;
use futures::future;
use rdkafka::{error::KafkaError, util::AsyncRuntime};
use rdkafka_sys::RDKafkaErrorCode;
use std::time::{Duration, Instant};

const KAFKA_CONNECT_TIMEOUT: Duration = Duration::from_secs(1);

pub struct SmolRuntime;

impl AsyncRuntime for SmolRuntime {
    type Delay = future::Map<smol::Timer, fn(Instant)>;

    fn spawn<T>(task: T)
    where
        T: Future<Output = ()> + Send + 'static,
    {
        // This needs to be smol::spawn we can't use async_std::task::spawn
        smol::spawn(task).detach();
    }

    fn delay_for(duration: Duration) -> Self::Delay {
        // This needs to be smol::Timer we can't use async_io::Timer
        futures::FutureExt::map(smol::Timer::after(duration), |_| ())
    }
}

/// verify broker host:port pairs in kafka connector configs
fn verify_brokers(id: &str, brokers: &[String]) -> Result<(String, Option<u16>)> {
    let mut first_broker: Option<(String, Option<u16>)> = None;
    for broker in brokers {
        match broker.split(':').collect::<Vec<_>>().as_slice() {
            [host] => {
                first_broker.get_or_insert_with(|| ((*host).to_string(), None));
            }
            [host, port] => {
                let port: u16 = port.parse().map_err(|_| {
                    Error::from(ErrorKind::InvalidConnectorDefinition(
                        id.to_string(),
                        format!("Invalid broker: {}:{}", host, port),
                    ))
                })?;
                first_broker.get_or_insert_with(|| ((*host).to_string(), Some(port)));
            }
            b => {
                return Err(ErrorKind::InvalidConnectorDefinition(
                    id.to_string(),
                    format!("Invalid broker: {}", b.join(":")),
                )
                .into())
            }
        }
    }
    first_broker.ok_or_else(|| {
        ErrorKind::InvalidConnectorDefinition(id.to_string(), "Missing brokers.".to_string()).into()
    })
}

/// Returns `true` if the error denotes a failed connect attempt
/// for both consumer and producer
fn is_failed_connect_error(err: &KafkaError) -> bool {
    matches!(
        err,
        KafkaError::ClientConfig(_, _, _, _)
            | KafkaError::ClientCreation(_)
            // TODO: what else?
            | KafkaError::Global(RDKafkaErrorCode::UnknownTopicOrPartition | RDKafkaErrorCode::UnknownTopic | RDKafkaErrorCode::AllBrokersDown)
    )
}

1	// Copyright 2022, The Tremor Team
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14	pub(crate) mod consumer;
15	pub(crate) mod producer;
16
17	use crate::errors::{Error, Kind as ErrorKind, Result};
18	use core::future::Future;
19	use futures::future;
20	use rdkafka::{error::KafkaError, util::AsyncRuntime};
21	use rdkafka_sys::RDKafkaErrorCode;
22	use std::time::{Duration, Instant};
23
24	const KAFKA_CONNECT_TIMEOUT: Duration = Duration::from_secs(1);
25
26	pub struct SmolRuntime;
27
28	impl AsyncRuntime for SmolRuntime {
29	type Delay = future::Map<smol::Timer, fn(Instant)>;
30
31	fn spawn<T>(task: T)	9✔
32	where	9✔
33	T: Future<Output = ()> + Send + 'static,	9✔
34	{	9✔
35	// This needs to be smol::spawn we can't use async_std::task::spawn	9✔
36	smol::spawn(task).detach();	9✔
37	}	9✔
38
39	fn delay_for(duration: Duration) -> Self::Delay {	9✔
40	// This needs to be smol::Timer we can't use async_io::Timer	9✔
41	futures::FutureExt::map(smol::Timer::after(duration), \|_\| ())	9✔
42	}	9✔
43	}
44
45	/// verify broker host:port pairs in kafka connector configs
46	fn verify_brokers(id: &str, brokers: &[String]) -> Result<(String, Option<u16>)> {	6✔
47	let mut first_broker: Option<(String, Option<u16>)> = None;	6✔
48	for broker in brokers {	12✔
49	match broker.split(':').collect::<Vec<_>>().as_slice() {	6✔
50	[host] => {	×
51	first_broker.get_or_insert_with(\|\| ((*host).to_string(), None));	×
52	}	×
53	[host, port] => {	6✔
54	let port: u16 = port.parse().map_err(\|_\| {	6✔
55	Error::from(ErrorKind::InvalidConnectorDefinition(	×
56	id.to_string(),	×
57	format!("Invalid broker: {}:{}", host, port),	×
58	))	×
59	})?;	6✔
60	first_broker.get_or_insert_with(\|\| ((*host).to_string(), Some(port)));	6✔
61	}
62	b => {	×
63	return Err(ErrorKind::InvalidConnectorDefinition(	×
64	id.to_string(),	×
65	format!("Invalid broker: {}", b.join(":")),	×
66	)	×
67	.into())	×
68	}
69	}
70	}
71	first_broker.ok_or_else(\|\| {	6✔
72	ErrorKind::InvalidConnectorDefinition(id.to_string(), "Missing brokers.".to_string()).into()	×
73	})	6✔
74	}	6✔
75
76	/// Returns `true` if the error denotes a failed connect attempt
77	/// for both consumer and producer
78	fn is_failed_connect_error(err: &KafkaError) -> bool {	11✔
79	matches!(	6✔
80	err,	11✔
81	KafkaError::ClientConfig(_, _, _, _)
82	\| KafkaError::ClientCreation(_)
83	// TODO: what else?
84	\| KafkaError::Global(RDKafkaErrorCode::UnknownTopicOrPartition \| RDKafkaErrorCode::UnknownTopic \| RDKafkaErrorCode::AllBrokersDown)
85	)
86	}	11✔

tremor-rs / tremor-runtime / 2270814456

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous