From a00d1de4414bb5ab57a27cb463af0e33af6e36c8 Mon Sep 17 00:00:00 2001 From: Stefan Miklosovic Date: Thu, 22 Jun 2023 14:23:49 +0200 Subject: [PATCH] Pass taskId from CompactionTask to system.compaction_history patch by Stefan Miklosovic; reviewed by Brandon Williams for CASSANDRA-12183 --- CHANGES.txt | 1 + src/java/org/apache/cassandra/db/SystemKeyspace.java | 5 +++-- .../cassandra/db/compaction/CompactionTask.java | 12 +++++++----- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 43b4cc40fb..fd862d225b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.30 + * Pass taskId from CompactionTask to system.compaction_history (CASSANDRA-12183) * Suppress CVE-2023-34455, CVE-2023-34454, CVE-2023-34453 (CASSANDRA-18608) * Backport CASSANDRA-10508: Remove hard-coded SSL cipher suites (CASSANDRA-18575) * Suppress CVE-2023-2976 (CASSANDRA-18562) diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index c0bf31779b..d2c3ff78ce 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -552,7 +552,8 @@ public final class SystemKeyspace SystemKeyspace.getOrInitializeLocalHostId(nodeIdSupplier); } - public static void updateCompactionHistory(String ksname, + public static void updateCompactionHistory(UUID taskId, + String ksname, String cfname, long compactedAt, long bytesIn, @@ -564,7 +565,7 @@ public final class SystemKeyspace return; String req = "INSERT INTO system.%s (id, keyspace_name, columnfamily_name, compacted_at, bytes_in, bytes_out, rows_merged) VALUES (?, ?, ?, ?, ?, ?, ?)"; executeInternal(String.format(req, COMPACTION_HISTORY), - UUIDGen.getTimeUUID(), + taskId, ksname, cfname, ByteBufferUtil.bytes(compactedAt), diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java index d023cef526..edd935ceeb 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java @@ -43,6 +43,7 @@ import org.apache.cassandra.db.compaction.CompactionManager.CompactionExecutorSt import org.apache.cassandra.db.lifecycle.LifecycleTransaction; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.UUIDGen; import org.apache.cassandra.utils.concurrent.Refs; public class CompactionTask extends AbstractCompactionTask @@ -232,9 +233,10 @@ public class CompactionTask extends AbstractCompactionTask return; double mbps = dTime > 0 ? (double) endsize / (1024 * 1024) / ((double) dTime / 1000) : 0; - Summary mergeSummary = updateCompactionHistory(cfs.keyspace.getName(), cfs.getColumnFamilyName(), mergedRowCounts, startsize, endsize); - logger.debug(String.format("Compacted (%s) %d sstables to [%s] to level=%d. %,d bytes to %,d (~%d%% of original) in %,dms = %fMB/s. %,d total partitions merged to %,d. Partition merge counts were {%s}", - taskId, transaction.originals().size(), newSSTableNames.toString(), getLevel(), startsize, endsize, (int) (ratio * 100), dTime, mbps, mergeSummary.totalSourceRows, totalKeysWritten, mergeSummary.partitionMerge)); + UUID taskTimeUUID = UUIDGen.getTimeUUID(); + Summary mergeSummary = updateCompactionHistory(taskTimeUUID, cfs.keyspace.getName(), cfs.getColumnFamilyName(), mergedRowCounts, startsize, endsize); + logger.debug(String.format("Compacted (task id: %s, compaction history id: %s) %d sstables to [%s] to level=%d. %,d bytes to %,d (~%d%% of original) in %,dms = %fMB/s. %,d total partitions merged to %,d. Partition merge counts were {%s}", + taskId, taskTimeUUID, transaction.originals().size(), newSSTableNames.toString(), getLevel(), startsize, endsize, (int) (ratio * 100), dTime, mbps, mergeSummary.totalSourceRows, totalKeysWritten, mergeSummary.partitionMerge)); logger.trace(String.format("CF Total Bytes Compacted: %,d", CompactionTask.addToTotalBytesCompacted(endsize))); logger.trace("Actual #keys: {}, Estimated #keys:{}, Err%: {}", totalKeysWritten, estimatedKeys, ((double)(totalKeysWritten - estimatedKeys)/totalKeysWritten)); } @@ -249,7 +251,7 @@ public class CompactionTask extends AbstractCompactionTask return new DefaultCompactionWriter(cfs, directories, transaction, nonExpiredSSTables, offline, keepOriginals); } - public static Summary updateCompactionHistory(String keyspaceName, String columnFamilyName, long[] mergedRowCounts, long startSize, long endSize) + public static Summary updateCompactionHistory(UUID taskTimeUUID, String keyspaceName, String columnFamilyName, long[] mergedRowCounts, long startSize, long endSize) { StringBuilder mergeSummary = new StringBuilder(mergedRowCounts.length * 10); Map mergedRows = new HashMap<>(); @@ -265,7 +267,7 @@ public class CompactionTask extends AbstractCompactionTask mergeSummary.append(String.format("%d:%d, ", rows, count)); mergedRows.put(rows, count); } - SystemKeyspace.updateCompactionHistory(keyspaceName, columnFamilyName, System.currentTimeMillis(), startSize, endSize, mergedRows); + SystemKeyspace.updateCompactionHistory(taskTimeUUID, keyspaceName, columnFamilyName, System.currentTimeMillis(), startSize, endSize, mergedRows); return new Summary(mergeSummary.toString(), totalSourceRows); }