Merge pull request #27 from aiven/dmitry-potepalov-rewrite-repaired-at

Reset repairedAt when receiving sstables
aiven · Nov 14, 2024 · 87e11bf · 87e11bf
2 parents 516b245 + 925fe27
commit 87e11bf
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 4 deletions.
diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraCompressedStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraCompressedStreamReader.java
@@ -75,7 +75,10 @@ public SSTableMultiWriter read(DataInputPlus inputPlus) throws Throwable
         try (CompressedInputStream cis = new CompressedInputStream(inputPlus, compressionInfo, ChecksumType.CRC32, cfs::getCrcCheckChance))
         {
             TrackedDataInputPlus in = new TrackedDataInputPlus(cis);
-            writer = createWriter(cfs, totalSize, repairedAt, pendingRepair, format);
+            if (repairedAt > 0) {
+                logger.info("[Stream #{}] Rewriting repairedAt from {} to 0 for sstable #{}", session.planId(), repairedAt, fileSeqNum);
+            }
+            writer = createWriter(cfs, totalSize, 0, pendingRepair, format);
             deserializer = new StreamDeserializer(cfs.metadata(), in, inputVersion, getHeader(cfs.metadata()), session, writer);
             String filename = writer.getFilename();
             String sectionName = filename + '-' + fileSeqNum;

diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java
@@ -134,10 +134,13 @@ public SSTableMultiWriter read(DataInputPlus in) throws Throwable
                              prettyPrintMemory(totalSize));
             }
 
+            if (messageHeader.repairedAt > 0) {
+                logger.info("[Stream #{}] Rewriting repairedAt from {} to 0 for sstable #{}", session.planId(), messageHeader.repairedAt, fileSequenceNumber);
+            }
             UnaryOperator<StatsMetadata> transform = stats -> stats.mutateLevel(header.sstableLevel)
-                                                                   .mutateRepairedMetadata(messageHeader.repairedAt, messageHeader.pendingRepair, false);
+                                                                   .mutateRepairedMetadata(0, messageHeader.pendingRepair, false);
             String description = String.format("level %s and repairedAt time %s and pendingRepair %s",
-                                               header.sstableLevel, messageHeader.repairedAt, messageHeader.pendingRepair);
+                                               header.sstableLevel, 0, messageHeader.pendingRepair);
             writer.descriptor.getMetadataSerializer().mutate(writer.descriptor, description, transform);
             return writer;
         }

diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java
@@ -135,7 +135,10 @@ public SSTableMultiWriter read(DataInputPlus inputPlus) throws Throwable
         try (StreamCompressionInputStream streamCompressionInputStream = new StreamCompressionInputStream(inputPlus, current_version))
         {
             TrackedDataInputPlus in = new TrackedDataInputPlus(streamCompressionInputStream);
-            writer = createWriter(cfs, totalSize, repairedAt, pendingRepair, format);
+            if (repairedAt > 0) {
+                logger.info("[Stream #{}] Rewriting repairedAt from {} to 0 for sstable #{}", session.planId(), repairedAt, fileSeqNum);
+            }
+            writer = createWriter(cfs, totalSize, 0, pendingRepair, format);
             deserializer = getDeserializer(cfs.metadata(), in, inputVersion, session, writer);
             String sequenceName = writer.getFilename() + '-' + fileSeqNum;
             long lastBytesRead = 0;