hubcio commented on code in PR #4738:
URL: https://github.com/apache/iggy/pull/4738#discussion_r4468747281
##########
core/integration/tests/cluster/crash_durability.rs:
##########
@@ -553,6 +617,65 @@ async fn
given_eager_flush_topic_when_the_whole_cluster_is_killed_should_recover
});
}
+/// The `replicated` twin of the eager-flush test above. A crash rolls a
+/// `replicated` partition back to the last WAL frontier it published, at a
+/// checkpoint, a durable operation or a graceful shutdown, so acks past it are
+/// lost even with every payload already in every node's segment files. Topics
+/// that must survive a whole-cluster crash use `persisted`.
+#[iggy_harness(cluster_nodes = 3)]
+async fn
given_replicated_eager_flush_topic_when_the_whole_cluster_is_killed_should_discard_acks_past_the_published_frontier(
+ harness: &mut TestHarness,
+) {
+ let client = harness.tcp_root_client().await.unwrap();
+ client
Review Comment:
simplification: this test repeats the existing stream and topic setup. add a
durability argument to `create_stream_and_topic` and reuse it here.
##########
core/integration/tests/cluster/crash_durability.rs:
##########
@@ -553,6 +617,65 @@ async fn
given_eager_flush_topic_when_the_whole_cluster_is_killed_should_recover
});
}
+/// The `replicated` twin of the eager-flush test above. A crash rolls a
+/// `replicated` partition back to the last WAL frontier it published, at a
+/// checkpoint, a durable operation or a graceful shutdown, so acks past it are
+/// lost even with every payload already in every node's segment files. Topics
+/// that must survive a whole-cluster crash use `persisted`.
+#[iggy_harness(cluster_nodes = 3)]
+async fn
given_replicated_eager_flush_topic_when_the_whole_cluster_is_killed_should_discard_acks_past_the_published_frontier(
+ harness: &mut TestHarness,
+) {
+ let client = harness.tcp_root_client().await.unwrap();
+ client
+ .create_stream(STREAM_NAME, &StreamCreateOptions::default())
+ .await
+ .expect("create stream");
+ client
+ .create_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
+ &TopicCreateOptions {
+ partitions_count: Some(1),
+ message_expiry: Some(IggyExpiry::NeverExpire),
+ messages_required_to_save: Some(1),
+ durability: Durability::Replicated,
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .expect("create topic");
+ let acked = produce_acked(&client, "flushed", 30).await;
+ let payloads: Vec<String> = acked.iter().map(|(_, payload)|
payload.clone()).collect();
+ wait_until_payloads_installed(harness, &payloads,
FLUSH_INSTALL_TIMEOUT).await;
+
+ // Same reason as above for keeping the producer's connection open.
+ harness.kill_cluster().expect("SIGKILL every node");
+ harness
+ .restart_cluster()
+ .await
+ .expect("restart the whole cluster");
+
+ let nodes: Vec<usize> = (0..harness.cluster_size()).collect();
+ let client = wait_until_cluster_serves(harness, &nodes,
CONVERGE_TIMEOUT).await;
+ // A write that reads back proves the partition serves again, so the
+ // missing acks below are discarded, not still recovering.
+ let after_restart = produce_acked_when_serving(&client,
CONVERGE_TIMEOUT).await;
+ wait_for_acked_readable(&client, &[after_restart], CONVERGE_TIMEOUT)
+ .await
+ .unwrap_or_else(|state| panic!("the partition should serve after the
restart: {state}"));
+ let readable = poll_payloads_from_start(&client, acked.len() as u32 +
16).await;
+ let survived = payloads
+ .iter()
+ .filter(|payload| readable.contains(payload))
+ .count();
+ assert!(
+ survived < acked.len(),
Review Comment:
warning: the test accepts partial recovery of unpublished acknowledgments,
so it does not enforce the stated rollback boundary. record the published
frontier before the crash and assert the exact surviving prefix afterward.
--
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]
