You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by GitBox <gi...@apache.org> on 2021/11/16 16:33:26 UTC
[GitHub] [cassandra] josh-mckenzie commented on a change in pull request #1311: CASSANDRA-17001: Optionally prune CDC segments if consumer fails to consume them fast enough
josh-mckenzie commented on a change in pull request #1311:
URL: https://github.com/apache/cassandra/pull/1311#discussion_r750417442
##########
File path: src/java/org/apache/cassandra/config/Config.java
##########
@@ -276,6 +276,9 @@
// Change-data-capture logs
public boolean cdc_enabled = false;
+ // When true, new CDC mutations are rejected/blocked when reaching max CDC storage.
+ // When false, new CDC mutations can always be added. But it will remove the oldest CDC commit log segment on full.
+ public volatile boolean cdc_block_writes = true;
Review comment:
(not strongly held opinion) - cdc_block_writes is ambiguous between cdc blocking write ack (like BatchCommitLogService) vs. blocking when space is full. Consider renaming to something like `cdc_block_when_full`
##########
File path: src/java/org/apache/cassandra/config/DatabaseDescriptor.java
##########
@@ -3096,6 +3096,17 @@ public static void setCDCEnabled(boolean cdc_enabled)
conf.cdc_enabled = cdc_enabled;
}
+ public static boolean shouldCDCBlockWrites()
Review comment:
Rename to `getCDCBlockWrites` to be consistent w/the rest of the file (I went through _a lot_ of this on the Denylist patch :) )
##########
File path: src/java/org/apache/cassandra/config/DatabaseDescriptor.java
##########
@@ -3096,6 +3096,17 @@ public static void setCDCEnabled(boolean cdc_enabled)
conf.cdc_enabled = cdc_enabled;
}
+ public static boolean shouldCDCBlockWrites()
+ {
+ return conf.cdc_block_writes;
+ }
+
+ public static void setCDCBlockWrites(boolean val)
+ {
+ Thread.dumpStack();
Review comment:
What's this dumpStack about? Left over from debugging?
##########
File path: src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDC.java
##########
@@ -193,23 +230,23 @@ public void addCDCSize(long size)
{
private final RateLimiter rateLimiter = RateLimiter.create(1000.0 / DatabaseDescriptor.getCDCDiskCheckInterval());
private ExecutorService cdcSizeCalculationExecutor;
- private CommitLogSegmentManagerCDC segmentManager;
-
- // Used instead of size during walk to remove chance of over-allocation
- private volatile long sizeInProgress = 0;
+ private final CommitLogSegmentManagerCDC segmentManager;
+ // track the total size between two dictionary size calculations
+ private final AtomicLong sizeInProgress;
Review comment:
I keep racking my brain trying to remember why I went w/a volatile initially instead of an Atomic, and I can't remember. I think the Atomic's a cleaner API anyway, so :+1: to this change.
##########
File path: src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java
##########
@@ -220,7 +223,20 @@ public String fileName()
public String cdcIndexFileName()
{
- return FILENAME_PREFIX + version + SEPARATOR + id + "_cdc.idx";
+ return FILENAME_PREFIX + version + SEPARATOR + id + INDEX_FILENAME_SUFFIX;
+ }
+
+ /**
+ * Infer the corresponding cdc index file using its cdc commitlog file
+ * @param cdcFile
+ * @return cdc index file
+ */
+ public static File inferCdcIndexFile(File cdcFile)
Review comment:
nit: consider renaming the variable `cdcCommitLogSegment` or something to disambiguate in the body of the method.
##########
File path: src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDC.java
##########
@@ -250,9 +302,11 @@ void processDiscardedSegment(CommitLogSegment segment)
{
// Add to flushed size before decrementing unflushed so we don't have a window of false generosity
if (segment.getCDCState() == CDCState.CONTAINS)
- size += segment.onDiskSize();
+ sizeInProgress.getAndAdd(segment.onDiskSize());
+
+ // Subtract the (estimated) size of the segment from processNewSegment.
if (segment.getCDCState() != CDCState.FORBIDDEN)
- size -= defaultSegmentSize();
+ sizeInProgress.getAndAdd(-defaultSegmentSize()); // todo: the defaultSegmentSize can be altered
Review comment:
Also, why is this `defaultSegmentSize` instead of just querying the `segment.onDiskSize()` like the call above? And yes, I know the simple answer is "that's because that's how you originally wrote it Josh", but I don't think this is... consistent?
##########
File path: src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDC.java
##########
@@ -78,11 +81,45 @@ public void discard(CommitLogSegment segment, boolean delete)
FileUtils.deleteWithConfirm(cdcLink);
File cdcIndexFile = segment.getCDCIndexFile();
- if (cdcIndexFile.exists())
- FileUtils.deleteWithConfirm(cdcIndexFile);
+ deleteCDCFiles(cdcLink, cdcIndexFile);
}
}
+ /**
+ * Delete the oldest CDC commit log segment to free up space.
+ * @return total deleted file size in bytes
+ */
+ public long deleteOldestCommitLogSegment()
Review comment:
Strongly recommend renaming this to `deleteOldestLinkedCDCSegmentFile` or something. It took me a double take of being Very Scared of this method before realizing it was deleting the oldest CDC segment and not just wildly deleting old CommitLogSegment files from the system. 😀
##########
File path: src/java/org/apache/cassandra/db/commitlog/CommitLog.java
##########
@@ -420,6 +421,28 @@ public long getActiveOnDiskSize()
return segmentRatios;
}
+ @Override
+ public boolean getCDCBlockWrites()
+ {
+ return DatabaseDescriptor.shouldCDCBlockWrites();
+ }
+
+ @Override
+ public void setCDCBlockWrites(boolean val)
+ {
+ Preconditions.checkState(DatabaseDescriptor.isCDCEnabled(),
+ "Unable to set block_writes (%s): CDC is not enabled.", val);
+ Preconditions.checkState(segmentManager instanceof CommitLogSegmentManagerCDC,
+ "Unexpected commit log segment manager type: %s", segmentManager.getClass().getName());
Review comment:
Consider making this error log more descriptive. Something like "Logic error! CDC is enabled but we have the wrong CommitLogSegmentManager type. Please report this as a bug." + the type.
Which is, admittedly, not the way we usually do error messages in our system, but IMO it'd be much more useful for an operator. If we've reached this point w/a disjoint between config param and the SegmentManager instantiated, something is _very very wrong_
##########
File path: src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDC.java
##########
@@ -78,11 +81,45 @@ public void discard(CommitLogSegment segment, boolean delete)
FileUtils.deleteWithConfirm(cdcLink);
File cdcIndexFile = segment.getCDCIndexFile();
- if (cdcIndexFile.exists())
- FileUtils.deleteWithConfirm(cdcIndexFile);
+ deleteCDCFiles(cdcLink, cdcIndexFile);
}
}
+ /**
+ * Delete the oldest CDC commit log segment to free up space.
+ * @return total deleted file size in bytes
+ */
+ public long deleteOldestCommitLogSegment()
+ {
+ File cdcDir = new File(DatabaseDescriptor.getCDCLogLocation());
+ Preconditions.checkState(cdcDir.isDirectory(), "The CDC directory does not exist.");
+ File[] files = cdcDir.tryList(f -> CommitLogDescriptor.isValid(f.name()));
+ Preconditions.checkState(files != null && files.length > 0, "There should be at least 1 CDC commit log segment.");
+ List<File> sorted = Arrays.stream(files)
+ .sorted(Comparator.comparingLong(File::lastModified))
+ .collect(Collectors.toList());
+ File oldestCdcFile = sorted.get(0);
+ File cdcIndexFile = CommitLogDescriptor.inferCdcIndexFile(oldestCdcFile);
+ return deleteCDCFiles(oldestCdcFile, cdcIndexFile);
+ }
+
+ private long deleteCDCFiles(File cdcFile, File cdcIndexFile)
+ {
+ long total = 0;
+ if (cdcFile.exists())
+ {
+ total += cdcFile.length();
+ FileUtils.deleteWithConfirm(cdcFile);
Review comment:
Looks like this is deprecated as of CEP-10. Is there another call we should use here? (a cursory inspection didn't surface anything for me, but the method is definitely flagged as Deprecated...)
##########
File path: src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java
##########
@@ -220,7 +223,20 @@ public String fileName()
public String cdcIndexFileName()
{
- return FILENAME_PREFIX + version + SEPARATOR + id + "_cdc.idx";
+ return FILENAME_PREFIX + version + SEPARATOR + id + INDEX_FILENAME_SUFFIX;
+ }
+
+ /**
+ * Infer the corresponding cdc index file using its cdc commitlog file
+ * @param cdcFile
+ * @return cdc index file
+ */
+ public static File inferCdcIndexFile(File cdcFile)
+ {
+ Preconditions.checkArgument(isValid(cdcFile.name()), "Invalid commit log");
Review comment:
It looks like this can throw an exception up stack and leave us with an inability to allocate new commit log files, repeatedly failing if the oldest CommitLogSegment is invalid.
##########
File path: src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDC.java
##########
@@ -250,9 +302,11 @@ void processDiscardedSegment(CommitLogSegment segment)
{
// Add to flushed size before decrementing unflushed so we don't have a window of false generosity
if (segment.getCDCState() == CDCState.CONTAINS)
- size += segment.onDiskSize();
+ sizeInProgress.getAndAdd(segment.onDiskSize());
+
+ // Subtract the (estimated) size of the segment from processNewSegment.
if (segment.getCDCState() != CDCState.FORBIDDEN)
- size -= defaultSegmentSize();
+ sizeInProgress.getAndAdd(-defaultSegmentSize()); // todo: the defaultSegmentSize can be altered
Review comment:
It looks like this alteration only takes place in 4 unit tests at this point. Maybe we should take this opportunity to document the `DatabaseDescriptor.setCommitLogSegmentSize` method as being for testing only and point to its usage in `CommitLogSegmentManagerCDC` via javadoc link for future maintainers that may be thinking about changing that?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org