spetz commented on code in PR #4154:
URL: https://github.com/apache/iggy/pull/4154#discussion_r3999143044
##########
core/connectors/sinks/elasticsearch_sink/src/lib.rs:
##########
@@ -231,12 +232,13 @@ impl ElasticsearchSink {
}
}
+ documents_indexed = items.len() - errors;
let mut state = self.state.lock().await;
state.errors_count += errors;
- state.documents_indexed += items.len() - errors;
+ state.documents_indexed += documents_indexed;
}
- Ok(())
+ Ok(documents_indexed)
Review Comment:
`bulk_index_documents` now knows how many documents Elasticsearch rejected
but still returns `Ok`, so `consume` at line 362 returns `Ok(())` and the
runtime Acks the batch and counts every message as processed. A bulk response
of HTTP 200 where every item carries an `error` therefore reports full success
to the runtime. The only external signal is one `warn!` per rejected document
plus `state.errors_count`, which is surfaced once in the close log at line 369
and never reaches the runtime's `errors_total` or `/stats`. The new integration
test `given_rejected_document_when_indexing_should_report_zero_indexed` locks
this in by sending `[1,2]` and asserting only that the log says zero. Worth
deciding whether a batch with `documents_indexed == 0` and a non-empty input
should return a nonzero FFI status now that PR 4152 makes the runtime act on
one. As a smaller point on the same line, the operator-facing message at line
357 reads "Successfully indexed 0 documents".
##########
core/connectors/sinks/doris_sink/config.toml:
##########
@@ -21,7 +21,6 @@ enabled = true
version = 0
name = "Doris sink"
path = "../../target/release/libiggy_connector_doris_sink"
-plugin_config_format = "toml"
Review Comment:
Drops `plugin_config_format = "toml"`. That is a no-op, since `ConfigFormat`
defaults to `Toml` at `core/connectors/runtime/src/configs/connectors.rs:47`,
and it matches fourteen of the sixteen sink configs. `surrealdb_sink` still
declares it.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -172,6 +173,40 @@ impl QuickwitSink {
}
}
+ async fn wait_for_index(
Review Comment:
`wait_for_index` probes readiness by sending POST with an empty body to the
real `?commit=auto` ingest endpoint rather than a metadata endpoint. Zero
documents should be a no-op, and the test asserts the probe bodies are empty,
but it does put a live ingest call on every `open()`.
##########
core/connectors/sinks/s3_sink/config.toml:
##########
@@ -20,7 +20,7 @@ key = "s3"
enabled = true
version = 0
name = "S3 sink"
-path = "../../target/release/libiggy_connector_s3_sink"
+path = "target/release/libiggy_connector_s3_sink"
Review Comment:
Same in `core/connectors/sinks/surrealdb_sink/config.toml:23` change `path`
from `../../target/release/...` to `target/release/...`. Both forms work:
`resolve_plugin_path` in `core/connectors/runtime/src/main.rs:349` tries the
literal path relative to the CWD first, then falls back to matching the file
name against the exe directory, which is where `../../target/...` gets rescued
from. The change is correct, but it leaves the tree at three sinks on the new
form (clickhouse was already there) and thirteen on the old one, against the
connectors-overview rule that plugin configs stay uniform. Either sweep the
rest or leave these two alone.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
