mlevkov commented on code in PR #4064:
URL: https://github.com/apache/iggy/pull/4064#discussion_r3997340497
##########
core/integration/tests/connectors/random/random_source.rs:
##########
@@ -97,6 +97,79 @@ async fn
state_save_failure_preserves_state_and_source_recovers(harness: &TestHa
random_source_liveness::assert_produces_messages(harness).await;
}
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/random/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sources_running_does_not_climb_across_restarts(harness: &TestHarness)
{
+ // The gauge counts running instances, and a restart takes one down before
+ // bringing one up, so one configured source stays at one however often it
+ // is restarted. It used to be reported by two mechanisms and taken back by
+ // one.
+ let api_url = harness
+ .connectors_runtime()
+ .expect("connectors runtime")
+ .http_url();
+ let http = Client::new();
+
+ wait_for_sources_running(&http, &api_url, 1).await;
+
+ for round in 1..=3 {
+ let response = http
+ .post(format!("{api_url}/sources/{SOURCE_KEY}/restart"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .expect("restart request should be sent");
+ assert_eq!(
+ response.status().as_u16(),
+ 204,
+ "restart {round} should be accepted"
+ );
+
+ wait_for_sources_running(&http, &api_url, 1).await;
+ }
+
+ // Read it once more after the gauge has settled. A report that lands after
+ // the poll above would otherwise go unseen.
+ sleep(STATE_STABILITY_WINDOW).await;
+ assert_eq!(
+ sources_running(&http, &api_url).await,
+ 1,
+ "one running source must stay counted once, whatever it took to
restart it"
+ );
+}
+
+async fn sources_running(http: &Client, api_url: &str) -> u32 {
Review Comment:
Done in `test(connectors): stop the runtime stats waits from hanging on a
stalled runtime`.
`fetch_stats` is the one request and the one decode, and it hands back
`reqwest::Result` exactly as you described, so `wait_for_source_error_after`
keeps treating a failed read as "not yet" while `sources_running` and
`source_errors` keep failing on it.
##########
core/connectors/runtime/src/manager/source.rs:
##########
@@ -463,6 +532,122 @@ mod tests {
assert_eq!(metrics.get_sources_running(), 1);
}
+ #[tokio::test]
+ async fn record_started_should_store_the_id_and_the_spawned_tasks() {
+ // Both have to land under one lock hold, so they are recorded together
+ // and there is nowhere to await between them. A stop reaches the
+ // instance through the id and drains it through the tasks, so losing
+ // either leaves something behind.
+ let mut details = create_test_source_details("pg", 1);
+ let config = details.config.clone();
+
+ details.record_started(7, &config, || vec![tokio::spawn(async {})]);
+
+ assert_eq!(
+ details.info.id, 7,
+ "a later stop closes whatever id this recorded"
+ );
+ assert_eq!(
+ details.handler_tasks.len(),
+ 1,
+ "a stop drains the tasks recorded here, so they cannot be dropped"
+ );
+ }
+
+ #[tokio::test]
+ async fn
should_not_double_count_when_an_error_falls_between_two_running_reports() {
+ // The interleaving spetz measured on #4064: the forwarding loop
reports
Review Comment:
Fixed in `refactor(connectors): type the cleanup label and say the argument
once`. It states the interleaving now, with no name and no PR number.
Fair catch, and it was our own rule: CLAUDE.md says comments explain why and
never reference the current task or PR. I wrote that one.
##########
core/connectors/runtime/src/main.rs:
##########
@@ -456,8 +456,28 @@ struct SinkConnectorWrapper {
plugins: Vec<SinkConnectorPlugin>,
}
+/// Closes a plugin instance whose setup did not finish, reporting a refusal
+/// rather than returning it: every caller is already on a failure path with an
+/// error of its own to surface.
+///
+/// `kind` is "source" or "sink". The two sides had this body inline, one word
+/// apart.
+pub(crate) fn close_plugin_instance(
+ close: &dyn Fn(u32) -> i32,
+ kind: &str,
Review Comment:
Done in `refactor(connectors): type the cleanup label and say the argument
once`. `close_plugin_instance` takes `ConnectorType` and `as_label` is
`pub(crate)`.
The old doc comment stated the constraint the signature did not enforce,
which is the tell.
##########
core/connectors/runtime/src/main.rs:
##########
@@ -456,8 +456,28 @@ struct SinkConnectorWrapper {
plugins: Vec<SinkConnectorPlugin>,
}
+/// Closes a plugin instance whose setup did not finish, reporting a refusal
+/// rather than returning it: every caller is already on a failure path with an
+/// error of its own to surface.
+///
+/// `kind` is "source" or "sink". The two sides had this body inline, one word
+/// apart.
+pub(crate) fn close_plugin_instance(
Review Comment:
Moved in `refactor(connectors): type the cleanup label and say the argument
once`. It sits above the grouped connector types now instead of between two of
them.
##########
core/integration/tests/connectors/random/random_source.rs:
##########
@@ -97,6 +97,79 @@ async fn
state_save_failure_preserves_state_and_source_recovers(harness: &TestHa
random_source_liveness::assert_produces_messages(harness).await;
}
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/random/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sources_running_does_not_climb_across_restarts(harness: &TestHarness)
{
+ // The gauge counts running instances, and a restart takes one down before
+ // bringing one up, so one configured source stays at one however often it
+ // is restarted. It used to be reported by two mechanisms and taken back by
+ // one.
+ let api_url = harness
+ .connectors_runtime()
+ .expect("connectors runtime")
+ .http_url();
+ let http = Client::new();
+
+ wait_for_sources_running(&http, &api_url, 1).await;
+
+ for round in 1..=3 {
+ let response = http
+ .post(format!("{api_url}/sources/{SOURCE_KEY}/restart"))
+ .header("api-key", API_KEY)
+ .send()
+ .await
+ .expect("restart request should be sent");
+ assert_eq!(
+ response.status().as_u16(),
+ 204,
+ "restart {round} should be accepted"
+ );
+
+ wait_for_sources_running(&http, &api_url, 1).await;
+ }
+
+ // Read it once more after the gauge has settled. A report that lands after
+ // the poll above would otherwise go unseen.
+ sleep(STATE_STABILITY_WINDOW).await;
Review Comment:
Renamed in `test(connectors): stop the runtime stats waits from hanging on a
stalled runtime`.
Took the shared option: both waits are waiting on the same thing, a report
that may land just after the poll before it, so it is `SETTLE_WINDOW` and the
doc comment says that is why they share it.
--
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]
