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.

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

@@ -136,13 +136,6 @@ final class TranslogHeader {
             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
             final long primaryTerm;
             if (version == VERSION_PRIMARY_TERM) {
@@ -160,6 +153,16 @@ final class TranslogHeader {
             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

@@ -38,6 +38,7 @@ import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.not;
 
 public class TranslogHeaderTests extends ESTestCase {
 
@@ -66,9 +67,10 @@ public class TranslogHeaderTests extends ESTestCase {
         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)) {
-                final TranslogHeader translogHeader = TranslogHeader.read(outHeader.getTranslogUUID(), translogFile, channel);
+                final TranslogHeader translogHeader = TranslogHeader.read(
+                    randomBoolean() ? outHeader.getTranslogUUID() : UUIDs.randomBase64UUID(), translogFile, channel);
                 // succeeds if the corruption corrupted the version byte making this look like a v2 translog, because we don't check the
                 // checksum on this version
                 assertThat("version " + TranslogHeader.VERSION_CHECKPOINTS + " translog",
@@ -82,6 +84,7 @@ public class TranslogHeaderTests extends ESTestCase {
                 throw new TranslogCorruptedException(translogFile.toString(), "adjusted translog version", e);
             }
         });
+        assertThat(corruption.getMessage(), not(containsString("this translog file belongs to a different translog")));
     }
 
     public void testHeaderWithoutPrimaryTerm() throws Exception {