This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch connectors_sink_delivery
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/connectors_sink_delivery by
this push:
new bd4dc3eec address review
bd4dc3eec is described below
commit bd4dc3eecd19f22f6a90974e2862592095974d1d
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Wed Nov 4 16:16:15 2026 +0100
address review
---
core/connectors/runtime/src/sink.rs | 7 ++--
core/connectors/sinks/http_sink/README.md | 1 +
core/connectors/sinks/http_sink/src/lib.rs | 55 +++++++++++++++++++++++-------
3 files changed, 47 insertions(+), 16 deletions(-)
diff --git a/core/connectors/runtime/src/sink.rs
b/core/connectors/runtime/src/sink.rs
index 60177fa35..e6033e382 100644
--- a/core/connectors/runtime/src/sink.rs
+++ b/core/connectors/runtime/src/sink.rs
@@ -364,8 +364,9 @@ pub(crate) fn spawn_consume_tasks(
.set_error(&plugin_key, &error.to_string(), Some(&metrics))
.await;
}
- // The polls also committed the batches the consumer still holds,
- // and only a shutdown moves the offsets back over them.
+ // At most once, the polls also committed the batches the consumer
+ // still holds, and only a shutdown moves the offsets back over
them.
+ // At least once, the polls store no offset.
if let Err(error) = consumer.shutdown().await {
error!(
"Failed to shut down the consumer of sink connector with
ID: {plugin_id}: {error}"
@@ -1026,7 +1027,7 @@ pub(crate) async fn process_messages(
Ok(next) => current_message = next,
Err(error) => {
error!(
- "Transform '{:?}' failed for sink connector with ID:
{plugin_id}, stream: {}, topic: {}: {error}",
+ "Transform '{:?}' failed for sink connector with ID:
{plugin_id}, stream: {}, topic: {}, offset: {message_offset}: {error}",
transform.r#type(),
topic_metadata.stream,
topic_metadata.topic
diff --git a/core/connectors/sinks/http_sink/README.md
b/core/connectors/sinks/http_sink/README.md
index b15c1142d..2570bf852 100644
--- a/core/connectors/sinks/http_sink/README.md
+++ b/core/connectors/sinks/http_sink/README.md
@@ -273,6 +273,7 @@ With the guarantee enabled:
- A message that `ndjson` or `json_array` mode would skip for a serialization
error fails the call instead.
- `individual` and `raw` modes stop at the first message that fails, after the
retries above, and at a payload above `max_payload_size_bytes`, so a later
message is not sent before an earlier one is delivered.
+- An `ndjson` or `json_array` batch above `max_payload_size_bytes` stops the
sink, since sending the same batch again cannot make it smaller. Lower
`batch_length` or raise `max_payload_size_bytes`, then restart the sink.
- A failure after retries with a non-transient status, such as 400 or 404,
stops the sink. A transient one is retried by the runtime, which hands over
again every message of the partition since its last flush. Without
`flush_interval` that is the failed batch. With it, the batches accepted before
the failure go again too, although the receiver already has them.
- The receiver can therefore see a message more than once, and should
deduplicate by `iggy_stream`, `iggy_topic`, `iggy_partition_id` and
`iggy_offset`. `raw` mode and `include_metadata = false` send none of these
fields, so the receiver has no key to deduplicate by.
diff --git a/core/connectors/sinks/http_sink/src/lib.rs
b/core/connectors/sinks/http_sink/src/lib.rs
index 3f272a64a..81e691d88 100644
--- a/core/connectors/sinks/http_sink/src/lib.rs
+++ b/core/connectors/sinks/http_sink/src/lib.rs
@@ -829,7 +829,7 @@ impl HttpSink {
);
// Count all successfully-serialized messages as errors (skipped
already counted individually)
self.errors_count.fetch_add(count, Ordering::Relaxed);
- return Err(Error::HttpRequestFailed(format!(
+ return Err(Error::InvalidRecordValue(format!(
"NDJSON batch exceeds max size: {} bytes",
body.len()
)));
@@ -928,7 +928,7 @@ impl HttpSink {
);
// Count all successfully-serialized messages as errors (skipped
already counted individually)
self.errors_count.fetch_add(count, Ordering::Relaxed);
- return Err(Error::HttpRequestFailed(format!(
+ return Err(Error::InvalidRecordValue(format!(
"JSON array batch exceeds max size: {} bytes",
body.len()
)));
@@ -1223,11 +1223,14 @@ impl Sink for HttpSink {
/// mitigates this by aborting early, but a fail-succeed-fail pattern can
bypass it.
///
/// **Runtime note**: The FFI boundary in `sdk/src/sink.rs` maps
`consume()`'s `Result` to
- /// `i32` (0=ok, 1=err), but the runtime's `process_messages()` in
`runtime/src/sink.rs`
- /// logs and counts a nonzero code, records no processed messages for the
run that
- /// returned it, and continues polling. A batch that mixes payload
variants is split
- /// into one `consume()` call per run, so the other runs of that poll
still count as
- /// processed. Returning `Err` does not trigger a runtime retry.
+ /// an `i32` status: 0 for `Ok`, 2 for an error that `Error::is_permanent`
reports and 1 for
+ /// any other error. At most once, the runtime's `process_messages()` in
+ /// `runtime/src/sink.rs` logs and counts a nonzero status, records no
processed messages
+ /// for the run that returned it, and continues polling, so an `Err` is
never retried. A
+ /// batch that mixes payload variants is split into one `consume()` call
per run, so the
+ /// other runs of that poll still count as processed. At least once, a 1
hands the messages
+ /// to `consume()` again from the first one not yet flushed, after a
backoff, while a 2, or
+ /// a 1 past `retry_max_attempts`, stops the delivery of the topic.
async fn consume(
&self,
topic_metadata: &TopicMetadata,
@@ -2111,14 +2114,14 @@ mod tests {
}
}
- /// A raw-mode sink pointed at a closed port, so every request that it
sends fails at once
- /// and `send_attempts` counts the requests.
- async fn given_unreachable_raw_sink(at_least_once: bool) -> HttpSink {
+ /// A sink pointed at a closed port, so every request that it sends fails
at once and
+ /// `send_attempts` counts the requests.
+ async fn given_unreachable_sink(batch_mode: BatchMode, at_least_once:
bool) -> HttpSink {
let mut sink = HttpSink::new(
1,
HttpSinkConfig {
url: "http://127.0.0.1:9/ingest".to_string(),
- batch_mode: Some(BatchMode::Raw),
+ batch_mode: Some(batch_mode),
max_payload_size_bytes: Some(8),
max_retries: Some(1),
health_check_enabled: Some(false),
@@ -2135,7 +2138,7 @@ mod tests {
#[tokio::test]
async fn
given_at_least_once_when_a_payload_is_too_large_should_fail_before_sending_the_rest()
{
- let sink = given_unreachable_raw_sink(true).await;
+ let sink = given_unreachable_sink(BatchMode::Raw, true).await;
let result = sink
.consume(
@@ -2154,7 +2157,7 @@ mod tests {
#[tokio::test]
async fn
given_at_most_once_when_a_payload_is_too_large_should_skip_it_and_send_the_rest()
{
- let sink = given_unreachable_raw_sink(false).await;
+ let sink = given_unreachable_sink(BatchMode::Raw, false).await;
let result = sink
.consume(
@@ -2168,6 +2171,32 @@ mod tests {
assert_eq!(sink.send_attempts.load(Ordering::Relaxed), 1);
}
+ #[tokio::test]
+ async fn
given_a_batch_mode_when_the_batch_is_too_large_should_fail_permanently_without_sending()
+ {
+ for batch_mode in [BatchMode::NdJson, BatchMode::JsonArray] {
+ let sink = given_unreachable_sink(batch_mode, true).await;
+
+ let result = sink
+ .consume(
+ &given_topic_metadata(),
+ given_messages_metadata(),
+ vec![given_raw_message(0, 64)],
+ )
+ .await;
+
+ assert!(
+ matches!(&result, Err(error) if error.is_permanent()),
+ "{batch_mode}: got {result:?}"
+ );
+ assert_eq!(
+ sink.send_attempts.load(Ordering::Relaxed),
+ 0,
+ "{batch_mode}"
+ );
+ }
+ }
+
#[test]
fn
given_a_final_status_when_classified_should_be_transient_only_where_the_middleware_retries()
{
