You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/16 19:55:47 UTC
(iggy) 01/07: test(integration): account for managed reconnects in stale clients
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/sdk-partition-routing
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 4c0589431ea5c51b704e00e3a2cd5b356b806ebf
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 15 23:58:01 2026 +0200
test(integration): account for managed reconnects in stale clients
Automatic replica reconnects renewed deliberately stale sessions and
added a second client after restart. Disable reconnect only for the
stale fixtures and explicitly disconnect the replaced stats client.
Those fixtures keep their expiry, membership, and persistence
assertions.
The auto reconnection test keeps reconnect on, so each evicted link
binds the same session again and renews its lease. Expect the
consumer to keep its membership through the silent window. Run the
test on one node, where every eviction takes the only link down, and
on three.
---
.../data_integrity/verify_after_server_restart.rs | 8 +--
.../verify_consumer_group_partition_assignment.rs | 4 ++
.../stale_client_consumer_group_scenario.rs | 63 ++++++++++++++--------
3 files changed, 48 insertions(+), 27 deletions(-)
diff --git a/core/integration/tests/data_integrity/verify_after_server_restart.rs b/core/integration/tests/data_integrity/verify_after_server_restart.rs
index a9dece200..9ed57b6b6 100644
--- a/core/integration/tests/data_integrity/verify_after_server_restart.rs
+++ b/core/integration/tests/data_integrity/verify_after_server_restart.rs
@@ -91,6 +91,9 @@ async fn should_fill_data_and_verify_after_restart() {
let expected_clients_count = stats.clients_count;
let expected_consumer_groups_count = stats.consumer_groups_count;
+ // Keep this handle from reconnecting alongside the replacement after restart.
+ client.disconnect().await.unwrap();
+
// Restart server
harness.restart_server().await.unwrap();
let server_addr = harness.server().raw_tcp_addr().unwrap();
@@ -170,10 +173,7 @@ async fn should_fill_data_and_verify_after_restart() {
"pinned-consumer",
IggyByteSize::from(amount_of_data_to_process.as_bytes_u64() * 2),
);
- drop(client_after_restart);
-
- // Connect and login to server
- let client = harness.tcp_root_client().await.unwrap();
+ let client = client_after_restart;
// Save stats from the second server (should have double the data)
let stats = client.get_stats().await.unwrap();
diff --git a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
index cb660f026..fd42ee27b 100644
--- a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
+++ b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
@@ -71,6 +71,10 @@ async fn create_stale_tcp_client(server_addr: &str) -> IggyClient {
let config = TcpClientConfig {
server_address: server_addr.to_string(),
heartbeat_interval: NonZeroIggyDuration::from_str("1h").unwrap(),
+ reconnection: TcpClientReconnectionConfig {
+ enabled: false,
+ ..TcpClientReconnectionConfig::default()
+ },
nodelay: true,
..TcpClientConfig::default()
};
diff --git a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
index e649ed757..6787665c5 100644
--- a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
+++ b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
@@ -125,7 +125,8 @@ async fn given_cluster_restart_when_member_is_replaced_should_resume_consumption
async fn given_live_member_when_backup_restarts_should_preserve_membership(
harness: &mut integration::harness::TestHarness,
) {
- let consumer_client = create_client(&harness.server().raw_tcp_addr().unwrap(), "1h").await;
+ let consumer_client =
+ create_client(&harness.server().raw_tcp_addr().unwrap(), "1h", true).await;
setup_resources(&consumer_client, Durability::Persisted).await;
let stream = Identifier::named(STREAM_NAME).unwrap();
let topic = Identifier::named(TOPIC_NAME).unwrap();
@@ -169,7 +170,7 @@ async fn verify_replacement_after_restart(
restart: Restart,
) {
let server_addr = harness.server().raw_tcp_addr().unwrap();
- let original = create_client(&server_addr, "500ms").await;
+ let original = create_client(&server_addr, "500ms", true).await;
setup_resources(&original, Durability::Persisted).await;
let stream = Identifier::named(STREAM_NAME).unwrap();
let topic = Identifier::named(TOPIC_NAME).unwrap();
@@ -232,7 +233,7 @@ async fn verify_replacement_after_restart(
harness.restart_node(0).unwrap();
}
- let replacement = create_client(&server_addr, "500ms").await;
+ let replacement = create_client(&server_addr, "500ms", true).await;
replacement
.login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
.await
@@ -313,10 +314,14 @@ async fn verify_replacement_after_restart(
}
}
-async fn create_client(server_addr: &str, heartbeat_interval: &str) -> IggyClient {
+async fn create_client(server_addr: &str, heartbeat_interval: &str, reconnect: bool) -> IggyClient {
let config = TcpClientConfig {
server_address: server_addr.to_string(),
heartbeat_interval: NonZeroIggyDuration::from_str(heartbeat_interval).unwrap(),
+ reconnection: TcpClientReconnectionConfig {
+ enabled: reconnect,
+ ..TcpClientReconnectionConfig::default()
+ },
nodelay: true,
..TcpClientConfig::default()
};
@@ -405,11 +410,11 @@ async fn should_handle_stale_client_with_manual_reconnection(
) {
let server_addr = harness.server().raw_tcp_addr().unwrap();
- let setup_client = create_client(&server_addr, "500ms").await;
+ let setup_client = create_client(&server_addr, "500ms", true).await;
setup_resources(&setup_client, Durability::Persisted).await;
- // Client with 1h heartbeat will become stale
- let stale_client = create_client(&server_addr, "1h").await;
+ // Rebinding after eviction would renew the logical session instead of letting it expire.
+ let stale_client = create_client(&server_addr, "1h", false).await;
stale_client
.login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
.await
@@ -485,7 +490,7 @@ async fn should_handle_stale_client_with_manual_reconnection(
// Reconnect with new client
drop(stale_client);
- let new_client = create_client(&server_addr, "500ms").await;
+ let new_client = create_client(&server_addr, "500ms", true).await;
new_client
.login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
.await
@@ -526,8 +531,9 @@ async fn should_handle_stale_client_with_manual_reconnection(
.await;
}
-/// Tests that IggyConsumer automatically recovers after stale disconnect.
-#[iggy_harness(server(
+/// Tests that IggyConsumer automatically recovers after stale disconnect. On one
+/// node every eviction takes the client's only link down.
+#[iggy_harness(cluster_nodes = [1, 3], server(
heartbeat.enabled = true,
heartbeat.interval = "7s",
consumer_group.heartbeat_interval = "500ms",
@@ -537,7 +543,7 @@ async fn should_handle_stale_client_with_auto_reconnection(
) {
let server_addr = harness.server().raw_tcp_addr().unwrap();
- let setup_client = create_client(&server_addr, "500ms").await;
+ let setup_client = create_client(&server_addr, "500ms", true).await;
setup_resources(&setup_client, Durability::Persisted).await;
let consumer_client = create_reconnecting_client(&server_addr).await;
@@ -571,18 +577,29 @@ async fn should_handle_stale_client_with_auto_reconnection(
if let Some(Ok(_)) = poll_result {
messages_consumed += 1;
if messages_consumed == 5 {
- // Stay silent until the verifier evicts the connection and
- // the server ends the session, so the consumer must
- // reconnect, log in again and rejoin its group.
- integration::harness::wait_for_consumer_group_assignment(
- &setup_client,
- &Identifier::named(STREAM_NAME).unwrap(),
- &Identifier::named(TOPIC_NAME).unwrap(),
- &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
- 0,
- STALE_EVICTION_TIMEOUT,
- )
- .await;
+ // Stay silent past the bound in which a silent client loses
+ // its membership. The heartbeat verifier evicts the
+ // consumer's links, but each link reconnects and binds the
+ // same session again, which renews its lease. The setup
+ // client runs no heartbeat, so these reads also keep its own
+ // session alive.
+ let deadline = tokio::time::Instant::now() + STALE_EVICTION_TIMEOUT;
+ while tokio::time::Instant::now() < deadline {
+ let group = setup_client
+ .get_consumer_group(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
+ )
+ .await
+ .unwrap()
+ .expect("consumer group exists");
+ assert_eq!(
+ group.members_count, 1,
+ "the rebound session must keep its membership: {group:?}"
+ );
+ sleep(STALE_EVICTION_RETRY_INTERVAL).await;
+ }
}
}
}