Reset repairedAt when receiving sstables

We're not using incremental repairs, therefore we want Repaired At to
be set to 0 for all sstables. Some older services still have this value
set to non-zero values from the times when we were using incremental
repairs. This patch should make sure that any new node we bootstrap
has Repaired At set to 0 for all sstables.

This patch should be removed when we release the next minor version
after 4.1.7.

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;
         }

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;