Verify translog checksum before UUID check (#49394)

When opening a translog file, we check whether the UUID matches what we expect (the UUID
from the latest commit). The UUID check can in certain cases fail when the translog is
corrupted. This commit changes the ordering of the checks so that corruption is detected first.

Author: ywelsch

server/src/main/java/org/elasticsearch/index/translog/TranslogHeader.java

@@ -137,13 +137,6 @@ static TranslogHeader read(final String translogUUID, final Path path, final Fil
             final BytesRef uuid = new BytesRef(uuidLen);
             uuid.length = uuidLen;
             in.read(uuid.bytes, uuid.offset, uuid.length);
-            final BytesRef expectedUUID = new BytesRef(translogUUID);
-            if (uuid.bytesEquals(expectedUUID) == false) {
-                throw new TranslogCorruptedException(
-                    path.toString(),
-                    "expected shard UUID " + expectedUUID + " but got: " + uuid +
-                        " this translog file belongs to a different translog");
-            }
             // Read the primary term
             assert version == VERSION_PRIMARY_TERM;
             final long primaryTerm = in.readLong();
@@ -154,6 +147,16 @@ static TranslogHeader read(final String translogUUID, final Path path, final Fil
             final int headerSizeInBytes = headerSizeInBytes(version, uuid.length);
             assert channel.position() == headerSizeInBytes :
                 "Header is not fully read; header size [" + headerSizeInBytes + "], position [" + channel.position() + "]";
+
+            // verify UUID only after checksum, to ensure that UUID is not corrupted
+            final BytesRef expectedUUID = new BytesRef(translogUUID);
+            if (uuid.bytesEquals(expectedUUID) == false) {
+                throw new TranslogCorruptedException(
+                    path.toString(),
+                    "expected shard UUID " + expectedUUID + " but got: " + uuid +
+                        " this translog file belongs to a different translog");
+            }
+
             return new TranslogHeader(translogUUID, primaryTerm, headerSizeInBytes);
         } catch (EOFException e) {
             throw new TranslogCorruptedException(path.toString(), "translog header truncated", e);

server/src/test/java/org/elasticsearch/index/translog/TranslogHeaderTests.java

@@ -31,6 +31,7 @@
 import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
 
 public class TranslogHeaderTests extends ESTestCase {
 
@@ -59,9 +60,9 @@ public void testCurrentHeaderVersion() throws Exception {
         for (int i = 0; i < corruptions && Files.size(translogFile) > 0; i++) {
             TestTranslog.corruptFile(logger, random(), translogFile, false);
         }
-        expectThrows(TranslogCorruptedException.class, () -> {
+        final TranslogCorruptedException corruption = expectThrows(TranslogCorruptedException.class, () -> {
             try (FileChannel channel = FileChannel.open(translogFile, StandardOpenOption.READ)) {
-                TranslogHeader.read(outHeader.getTranslogUUID(), translogFile, channel);
+                TranslogHeader.read(randomBoolean() ? outHeader.getTranslogUUID() : UUIDs.randomBase64UUID(), translogFile, channel);
             } catch (IllegalStateException e) {
                 // corruption corrupted the version byte making this look like a v2, v1 or v0 translog
                 assertThat("version " + TranslogHeader.VERSION_CHECKPOINTS + "-or-earlier translog",
@@ -70,6 +71,7 @@ public void testCurrentHeaderVersion() throws Exception {
                 throw new TranslogCorruptedException(translogFile.toString(), "adjusted translog version", e);
             }
         });
+        assertThat(corruption.getMessage(), not(containsString("this translog file belongs to a different translog")));
     }
 
     public void testLegacyTranslogVersions() {