frankgh commented on code in PR #4892:
URL: https://github.com/apache/cassandra/pull/4892#discussion_r3437834856
##########
src/java/org/apache/cassandra/replication/MutationJournal.java:
##########
@@ -79,11 +85,27 @@
// TODO (required): handle table truncations
public class MutationJournal
{
+ private static final Logger logger =
LoggerFactory.getLogger(MutationJournal.class);
+
+ // opaque / immutable list of segments that we should clear the
needs-replay flag on
+ public static class PendingClearReplay
+ {
+ private ImmutableSet<Long> segments;
Review Comment:
NIT: can we make this final?
```suggestion
private final mmutableSet<Long> segments;
```
##########
src/java/org/apache/cassandra/metrics/MutationTrackingMetrics.java:
##########
@@ -63,5 +64,9 @@ private MutationTrackingMetrics()
factory.createMetricName("JournalDiskSpaceUsed"),
() -> MutationJournal.instance().getDiskSpaceUsed()
);
+ pendingClearReplaySize = Metrics.register(
+ factory.createMetricName("PendingClearReplaySize"),
+ () -> MutationJournal.instance().pendingClearReplaySize()
Review Comment:
Do we want to worry about the case where MT is disabled and maybe handle the
`IllegalStateException` thrown when the instance is null?
##########
src/java/org/apache/cassandra/replication/MutationTrackingService.java:
##########
Review Comment:
should we conditionally remove the listener here, only if the service was
started?
```suggestion
boolean wasStarted;
synchronized (this)
{
wasStarted = started;
if (wasStarted)
ClusterMetadataService.instance().log().removeListener(tcmListener);
}
```
##########
src/java/org/apache/cassandra/replication/MutationTrackingService.java:
##########
Review Comment:
additionally, we need to synchronize for access to the started volatile
variable.
##########
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.size() == 0)
Review Comment:
NIT:
```suggestion
if (summary.isEmpty())
```
##########
src/java/org/apache/cassandra/replication/MutationTrackingService.java:
##########
@@ -323,6 +337,10 @@ private void shutdownBlocking() throws InterruptedException
activeReconciler.shutdownBlocking();
executor.shutdown();
executor.awaitTermination(1, TimeUnit.MINUTES);
Review Comment:
Should we log if we fail to shutdown here?
```suggestion
if (!executor.awaitTermination(1, TimeUnit.MINUTES))
{
logger.warn("Mutation tracking executor did not terminate within
1 minute; forcing shutdown");
}
```
##########
src/java/org/apache/cassandra/replication/MutationTrackingService.java:
##########
@@ -323,6 +337,10 @@ private void shutdownBlocking() throws InterruptedException
activeReconciler.shutdownBlocking();
executor.shutdown();
executor.awaitTermination(1, TimeUnit.MINUTES);
+ // attempt to persist offsets and mark segments as
+ // not needing replay one last time before shutdown
+ if (started)
Review Comment:
```suggestion
if (wasStarted)
```
##########
src/java/org/apache/cassandra/replication/MutationTrackingService.java:
##########
@@ -1576,27 +1602,82 @@ private void run(Shard shard, boolean durable)
}
}
+ /**
+ * Persists per-log witnessed offsets, and durably marks needsReplay=false
on any segments that have become eligible
+ * for it since the most recent run of this class. These 2 operations need
to performed in a specific sequence to avoid
+ * correctness problems.
+ *
+ * For background, mutation tracking needs to keep a record of every
mutation id it's written locally. For correctness
+ * purposes, a nodes view of mutation ids it's written locally needs to
exactly match the data it has on disk.
+ * Having data on disk you dont have an id for, or thinking you have ids
on disk that you don't breaks the mutation
+ * tracking consistency mechanism.
+ *
+ * To improve startup, we periodically save our view of mutation ids that
we've witnessed to disk as part of this
+ * class. Any ids witnessed since the last time this class was run are
reconstructed by replaying the journal.
+ *
+ * However, if an sstable is flushed is after the most recent
LogStatePersister run, AND it marks a segment as no
Review Comment:
NIT:
```suggestion
* However, if an sstable is flushed after the most recent
LogStatePersister run, AND it marks a segment as no
```
--
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]
