Go back to polling for the kafka onramp

CHANGELOG.md

@@ -6,6 +6,13 @@
 
 * Allow using '_' as seperators in numeric literals [#645](https://github.com/tremor-rs/tremor-runtime/issues/645)
 * Add support for Kafka message headers, available through the `$kafka_headers` metadata variable.
+* Add the `cb` offramp for testing upstream circuit breaker behaviour [#779](https://github.com/tremor-rs/tremor-runtime/pull/779)
+* Add the `kafka` onramp config `retry_failed_events` to acoid retrying failed events, and `polling_interval` to control how often kafka is polled for new messages if none were available previously [#779](https://github.com/tremor-rs/tremor-runtime/pull/779)
+
+### Fixes
+
+* Fix `kafka` onramp hanging with no message in the queue, leading to delayed offset commits [#779](https://github.com/tremor-rs/tremor-runtime/pull/779)
+* Fail the `kafka` onramp if any of the configured topics could not be subscribed to [#779](https://github.com/tremor-rs/tremor-runtime/pull/779)
 
 ## 0.10.2
 

Cargo.toml

@@ -32,7 +32,6 @@ anyhow = "1"
 async-channel = "1"
 async-compat = "0.2"
 async-compression = {version = "0.3", features = ["xz", "futures-bufread", "stream"]}
-async-io = "1.3"
 async-std = {version = "1.9.0", features = ["unstable", "attributes", "tokio03", "tokio1"]}
 async-trait = "0.1"
 async-tungstenite = {version = "0.13.0", features = ["async-std-runtime"]}
@@ -65,7 +64,6 @@ serde_derive = "1"
 serde_yaml = "0.8"
 simd-json = {version = "0.3", features = ["known-key"]}
 simd-json-derive = "0.1.15"
-smol = "1.2"
 snap = "1"
 surf = "=2.1.0"
 tremor-common = {path = "tremor-common"}

src/lib.rs

@@ -29,8 +29,6 @@ extern crate serde_derive;
 extern crate log;
 #[macro_use]
 extern crate lazy_static;
-#[macro_use]
-extern crate rental;
 
 #[macro_use]
 pub(crate) mod macros;

src/offramp.rs

@@ -19,8 +19,8 @@ use crate::permge::PriorityMerge;
 use crate::pipeline;
 use crate::registry::ServantId;
 use crate::sink::{
-    self, blackhole, debug, elastic, exit, file, handle_response, kafka, newrelic, postgres, rest,
-    stderr, stdout, tcp, udp, ws,
+    self, blackhole, cb, debug, elastic, exit, file, handle_response, kafka, newrelic, postgres,
+    rest, stderr, stdout, tcp, udp, ws,
 };
 use crate::source::Processors;
 use crate::url::ports::{IN, METRICS};
@@ -110,6 +110,7 @@ pub fn lookup(name: &str, config: &Option<OpConfig>) -> Result<Box<dyn Offramp>>
         "debug" => debug::Debug::from_config(config),
         "elastic" => elastic::Elastic::from_config(config),
         "exit" => exit::Exit::from_config(config),
+        "cb" => cb::CB::from_config(config),
         "file" => file::File::from_config(config),
         "kafka" => kafka::Kafka::from_config(config),
         "newrelic" => newrelic::NewRelic::from_config(config),

src/sink.rs

@@ -20,6 +20,7 @@ use async_channel::Sender;
 use halfbrown::HashMap;
 
 pub(crate) mod blackhole;
+pub(crate) mod cb;
 pub(crate) mod debug;
 pub(crate) mod elastic;
 pub(crate) mod exit;