frankgh commented on code in PR #4892:
URL: https://github.com/apache/cassandra/pull/4892#discussion_r3452810618
##########
test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingTest.java:
##########
@@ -107,6 +109,260 @@ public void testBasicWritePath() throws Throwable
}
}
+ private static int getOffsetCount(IInvokableInstance node, String
keyspaceName, String tableName, int key)
+ {
+ return node.callOnInstance(() -> {
+ TableMetadata table =
Schema.instance.getTableMetadata(keyspaceName, tableName);
+ DecoratedKey dk =
Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(key));
+ MutationSummary summary =
MutationTrackingService.instance().createSummaryForKey(dk, table.id, false);
+ if (summary.isEmpty())
+ return 0;
+ CoordinatorLogId logId = getOnlyLogId(summary);
+ return summaryIdSpace(summary.get(logId)).offsetCount();
+ });
+
+ }
+
+ private static int getOffsetCount(IInvokableInstance node, String
keyspaceName, int key)
+ {
+ return getOffsetCount(node, keyspaceName, "tbl", key);
+ }
+
+ /**
+ * Writes tracked mutations, deliberately doesn't flush so the writes live
only in the
+ * commit log the node, and asserts MTS witness state on boot reflects the
unflushed writes. Confirm they're
+ * reconstructed on journal playback
+ */
+ @Test
+ public void testWitnessSurvivesBounceWithoutFlush() throws Throwable
+ {
+ final int key = 1;
+ final int writes = 10;
+
+ try (Cluster cluster = Cluster.build(1)
+ .withConfig(cfg ->
cfg.with(Feature.NETWORK).with(Feature.GOSSIP))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH
replication = " +
+ "{'class': 'SimpleStrategy',
'replication_factor': 1} " +
+ "AND
replication_type='tracked';"));
+ cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (k int
PRIMARY KEY, v int);"));
+
+ // Pause the persister so writes never reach
system.coordinator_logs SSTables —
+ // the only durable record of the witnesses on disk lives in the
commit log.
+ cluster.get(1).runOnInstance(() ->
MutationTrackingService.instance().pauseOffsetsPersisterForTesting());
+
+ for (int i = 0; i < writes; i++)
+ cluster.coordinator(1).execute(withKeyspace("INSERT INTO
%s.tbl (k, v) VALUES (?, ?)"),
+ ConsistencyLevel.QUORUM, key,
i);
+
+ String keyspaceName = KEYSPACE;
+ int preBounceOffsetCount = getOffsetCount(cluster.get(1),
keyspaceName, key);
+ assertEquals("Pre-bounce witness count must equal write count",
writes, preBounceOffsetCount);
+
+ // Bounce without flushing. Witnesses live only in the commit log
+ journal segments
+ // (still active, needsReplay=true). On the way back up, MTS.start
must run after
+ // CommitLog.recoverSegmentsOnDisk() so the journal replay path
repopulates witnesses
+ // before any consumer queries MTS state.
+ ClusterUtils.stopUnchecked(cluster.get(1));
+ cluster.get(1).startup();
+
+ cluster.get(1).runOnInstance(() ->
MutationTrackingService.instance().pauseOffsetsPersisterForTesting());
+
+ int postBounceOffsetCount = getOffsetCount(cluster.get(1),
keyspaceName, key);
+
+ assertEquals("Witness state must survive bounce-without-flush:
post-bounce offsets must match pre-bounce",
+ preBounceOffsetCount, postBounceOffsetCount);
+ }
+ }
+
+ /**
+ * Regression test for the lost-witness-marker race (CASSANDRA-21443).
+ *
+ * When a memtable flush + segment close fires before the periodic
LogStatePersister
+ * has written witnessed offsets to system.coordinator_logs, the segment
metadata can be
+ * durably marked needsReplay=false while the witnesses for its mutations
are still only
+ * in memory. A crash in this window leaves the node with data in SSTables
but with
+ * witness state missing on restart, breaking mutation summaries and
journal sync barrier
+ * guarantees.
+ *
+ * The test pauses the persister, writes a known set of mutations, forces
flush and
+ * segment close (triggering maybeCleanupStaticSegment), bounces the node,
and asserts
+ * that the post-restart witness state matches the pre-bounce snapshot.
+ */
+ @Test
+ public void testWitnessSurvivesCrashAfterFlushAndSegmentClose() throws
Throwable
+ {
+ final int key = 1;
+ final int writes = 10;
+
+ try (Cluster cluster = Cluster.build(1)
+ .withConfig(cfg ->
cfg.with(Feature.NETWORK).with(Feature.GOSSIP))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH
replication = " +
+ "{'class': 'SimpleStrategy',
'replication_factor': 1} " +
+ "AND
replication_type='tracked';"));
+ cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (k int
PRIMARY KEY, v int);"));
+
+ // Pause the persister so no witness state escapes to
system.coordinator_logs
+ // for the duration of the test window. This mirrors the
in-production hazard
+ // between persister ticks (currently 1s, planned 60s).
+ cluster.get(1).runOnInstance(() ->
MutationTrackingService.instance().pauseOffsetsPersisterForTesting());
+
+ for (int i = 0; i < writes; i++)
+ cluster.coordinator(1).execute(withKeyspace("INSERT INTO
%s.tbl (k, v) VALUES (?, ?)"),
+ ConsistencyLevel.QUORUM, key,
i);
+
+ String keyspaceName = KEYSPACE;
+ int preBounceOffsetCount = getOffsetCount(cluster.get(1),
keyspaceName, key);
+ assertEquals("Pre-bounce witness count must equal write count",
writes, preBounceOffsetCount);
+
+ // Flush so notifyFlushed marks the active segment's interval
clean.
+ cluster.get(1).nodetoolResult("flush",
KEYSPACE).asserts().success();
Review Comment:
should we flush `system.coordinator_logs` here as well?
```
cluster.get(1).nodetoolResult("flush", "system",
"coordinator_logs").asserts().success();
```
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
