[Java] Complete adding header and version information to the recording log.

mikeb01 · mikeb01 · commit b82ecfb7bd72 · 2024-09-03T08:45:49.000+12:00
diff --git a/aeron-cluster/src/main/java/io/aeron/cluster/RecordingLog.java b/aeron-cluster/src/main/java/io/aeron/cluster/RecordingLog.java
@@ -23,6 +23,7 @@
 import org.agrona.DirectBuffer;
 import org.agrona.LangUtil;
 import org.agrona.MutableDirectBuffer;
+import org.agrona.SemanticVersion;
 import org.agrona.Strings;
 import org.agrona.collections.IntArrayList;
 import org.agrona.collections.Long2LongHashMap;
@@ -50,6 +51,9 @@
 import static io.aeron.archive.client.AeronArchive.NULL_POSITION;
 import static java.lang.Math.max;
 import static java.nio.ByteOrder.LITTLE_ENDIAN;
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.COPY_ATTRIBUTES;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
 import static java.nio.file.StandardOpenOption.*;
 import static org.agrona.BitUtil.*;
 
@@ -68,12 +72,13 @@
  *   0                   1                   2                   3
  *   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
  *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
- *  |                  0xFFA3F010 (Magic Number)                    |
- *  |                                                               |
+ *  |                        0xFFA3F010 (Magic Number)              |
+ *  |                        0x00000000                             |
  *  +---------------------------------------------------------------+
  *  |                          Version                              |
  *  +---------------------------------------------------------------+
- *  |                          Reserved                             |
+ *  |                      Reserved (52 bytes)                    ...
+ *  ...                                                             |
  *  +---------------------------------------------------------------+
  * </pre>
  * Recording log entry as follows:
@@ -117,7 +122,13 @@
 public final class RecordingLog implements AutoCloseable
 {
     public static final long MAGIC_NUMBER = 0xFFA3F010_00000000L;
-    private static final int HEADER_SIZE = 64;
+    public static final int HEADER_SIZE = 64;
+    public static final int MAJOR_VERSION = 0;
+    public static final int MINOR_VERSION = 1;
+    public static final int PATCH_VERSION = 0;
+    public static final int SEMANTIC_VERSION = SemanticVersion.compose(MAJOR_VERSION, MINOR_VERSION, PATCH_VERSION);
+    private static final int MAGIC_NUMBER_OFFSET = 0;
+    private static final int VERSION_OFFSET = MAGIC_NUMBER_OFFSET + SIZE_OF_LONG;
 
     /**
      * Representation of the entry in the {@link RecordingLog}.
@@ -646,6 +657,9 @@ public String toString()
      */
     public static final String RECORDING_LOG_FILE_NAME = "recording.log";
 
+    static final String RECORDING_LOG_MIGRATED_FILE_NAME = RECORDING_LOG_FILE_NAME + ".migrated";
+    static final String RECORDING_LOG_NEW_FILE_NAME = RECORDING_LOG_FILE_NAME + ".new";
+
     /**
      * The log entry is for a recording of messages within a leadership term to the log.
      */
@@ -785,6 +799,7 @@ public RecordingLog(final File parentDir, final boolean createNew)
             if (isNewFile)
             {
                 syncDirectory(parentDir);
+                writeHeader(fileChannel);
             }
             else
             {
@@ -802,54 +817,6 @@ public RecordingLog(final File parentDir, final boolean createNew)
         }
     }
 
-    private void checkForVersionAndMigrate(final File logFile) throws IOException
-    {
-        if (requiresMigration(logFile))
-        {
-            final File oldMigratedFile = new File(logFile.getParentFile(), RECORDING_LOG_FILE_NAME + ".migrated");
-            if (!logFile.renameTo(oldMigratedFile))
-            {
-                throw new IOException("Unable to backup old file to new one");
-            }
-
-            final File newFile = new File(logFile.getParentFile(), RECORDING_LOG_FILE_NAME);
-            final MutableDirectBuffer header = new UnsafeBuffer(new byte[HEADER_SIZE]);
-            // TODO: Offset constants
-            header.putLong(0, MAGIC_NUMBER, LITTLE_ENDIAN);
-            header.putInt(8, 0 /* TODO: version */, LITTLE_ENDIAN);
-
-            try (FileOutputStream outputStream = new FileOutputStream(newFile))
-            {
-                outputStream.write(header.byteArray());
-                Files.copy(oldMigratedFile.toPath(), outputStream);
-            }
-        }
-    }
-
-    private boolean requiresMigration(final File logFile) throws IOException
-    {
-        if (logFile.length() < HEADER_SIZE)
-        {
-            return true;
-        }
-
-        try (FileChannel fileChannel = FileChannel.open(logFile.toPath(), READ))
-        {
-            return isMagicNumberInvalid(fileChannel);
-        }
-    }
-
-    private static boolean isMagicNumberInvalid(final FileChannel fileChannel) throws IOException
-    {
-        final DirectBuffer header = new UnsafeBuffer(ByteBuffer.allocateDirect(HEADER_SIZE));
-        if (HEADER_SIZE != fileChannel.read(header.byteBuffer()))
-        {
-            throw new IOException("Unable to read header");
-        }
-        final long magicNumber = header.getLong(0, LITTLE_ENDIAN);
-        return magicNumber != MAGIC_NUMBER;
-    }
-
     /**
      * {@inheritDoc}
      */
@@ -1884,6 +1851,73 @@ private int captureEntriesFromBuffer(
         return consumed;
     }
 
+    private void applyHeader(final MutableDirectBuffer header)
+    {
+        header.putLong(MAGIC_NUMBER_OFFSET, MAGIC_NUMBER, LITTLE_ENDIAN);
+        header.putInt(VERSION_OFFSET, SEMANTIC_VERSION, LITTLE_ENDIAN);
+    }
+
+    private void writeHeader(final FileChannel fileChannel) throws IOException
+    {
+        final ByteBuffer headerBuffer = ByteBuffer.allocateDirect(HEADER_SIZE);
+        final MutableDirectBuffer header = new UnsafeBuffer(headerBuffer);
+        applyHeader(header);
+
+        if (HEADER_SIZE != fileChannel.write(headerBuffer))
+        {
+            throw new IOException("Failed to write full header");
+        }
+    }
+
+    private void checkForVersionAndMigrate(final File logFile) throws IOException
+    {
+        if (requiresMigration(logFile))
+        {
+            final File oldMigratedFile = new File(logFile.getParentFile(), RECORDING_LOG_MIGRATED_FILE_NAME);
+            Files.copy(logFile.toPath(), oldMigratedFile.toPath(), COPY_ATTRIBUTES, REPLACE_EXISTING);
+
+            final File newFile = new File(logFile.getParentFile(), RECORDING_LOG_NEW_FILE_NAME);
+            Files.deleteIfExists(newFile.toPath());
+
+            final MutableDirectBuffer header = new UnsafeBuffer(new byte[HEADER_SIZE]);
+            applyHeader(header);
+
+            try (FileOutputStream outputStream = new FileOutputStream(newFile, false))
+            {
+                outputStream.write(header.byteArray());
+                Files.copy(oldMigratedFile.toPath(), outputStream);
+            }
+
+            Files.move(
+                newFile.toPath(), new File(logFile.getParentFile(), RECORDING_LOG_FILE_NAME).toPath(),
+                ATOMIC_MOVE, REPLACE_EXISTING);
+        }
+    }
+
+    private boolean requiresMigration(final File logFile) throws IOException
+    {
+        if (logFile.length() < HEADER_SIZE)
+        {
+            return true;
+        }
+
+        try (FileChannel fileChannel = FileChannel.open(logFile.toPath(), READ))
+        {
+            return isMagicNumberInvalid(fileChannel);
+        }
+    }
+
+    private static boolean isMagicNumberInvalid(final FileChannel fileChannel) throws IOException
+    {
+        final DirectBuffer header = new UnsafeBuffer(ByteBuffer.allocateDirect(HEADER_SIZE));
+        if (HEADER_SIZE != fileChannel.read(header.byteBuffer()))
+        {
+            throw new IOException("Unable to read header");
+        }
+        final long magicNumber = header.getLong(0, LITTLE_ENDIAN);
+        return magicNumber != MAGIC_NUMBER;
+    }
+
     private static void syncDirectory(final File dir)
     {
         try (FileChannel fileChannel = FileChannel.open(dir.toPath()))
diff --git a/aeron-cluster/src/test/java/io/aeron/cluster/RecordingLogTest.java b/aeron-cluster/src/test/java/io/aeron/cluster/RecordingLogTest.java
@@ -670,6 +670,7 @@ void entriesInTheRecordingLogShouldBeSorted()
 
             recordingLog.reload();
 
+            assertEquals(sortedList.size(), recordingLog.entries().size());
             assertEquals(sortedList, recordingLog.entries()); // reload from disc and re-sort
         }
     }
diff --git a/aeron-cluster/src/test/java/io/aeron/cluster/RecordingLogVersioningTest.java b/aeron-cluster/src/test/java/io/aeron/cluster/RecordingLogVersioningTest.java
@@ -15,30 +15,34 @@
  */
 package io.aeron.cluster;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
-import java.nio.file.StandardCopyOption;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.Random;
 
+import static io.aeron.cluster.RecordingLog.ENTRY_TYPE_SNAPSHOT;
+import static io.aeron.cluster.RecordingLog.ENTRY_TYPE_STANDBY_SNAPSHOT;
 import static io.aeron.cluster.RecordingLog.ENTRY_TYPE_TERM;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class RecordingLogVersioningTest
 {
     private static final byte[] CHARACTER_TABLE = new byte[27];
     public static final long FIXED_SEED_FOR_CONSISTENT_DATA = 892374458763L;
 
+    @TempDir File tempDirA;
+    @TempDir File tempDirB;
+
     static
     {
         for (int i = 0; i < 26; i++)
@@ -60,7 +64,7 @@ private static class TestEntry
         private final int serviceId;
         private final String endpoint;
 
-        public TestEntry(
+        TestEntry(
             final int entryType,
             final long recordingId,
             final long leadershipTermId,
@@ -140,13 +144,19 @@ private List<TestEntry> generateData()
         return entries;
     }
 
-    @SuppressWarnings("checkstyle:MethodName")
     @Test
-    @Disabled // Used to generate existing data.
-    void generateRecordingLogForVersionTest_1_45()
+    void verifyDataRemainsConsistent()
+    {
+        assertEquals(generateData(), generateData());
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = { true, false })
+    void shouldLoadOldVersionAndMigrate(final boolean hasPartiallyProcessedFiles) throws IOException
     {
-        final File parentDir = new File("src/test/resources/v1_45_x");
-        try (UnversionedRecordingLog recordingLog = new UnversionedRecordingLog(parentDir, true))
+        assertNotEquals(tempDirA, tempDirB);
+
+        try (UnversionedRecordingLog recordingLog = new UnversionedRecordingLog(tempDirA, true))
         {
             for (final TestEntry testEntry : generateData())
             {
@@ -160,17 +170,20 @@ void generateRecordingLogForVersionTest_1_45()
                     testEntry.serviceId,
                     testEntry.endpoint);
             }
-            final List<UnversionedRecordingLog.Entry> entries = recordingLog.entries();
-            final String s = entries.toString();
-            System.out.println(s);
         }
-    }
 
-    private static void appendToRecordingLog(final RecordingLog recordingLog, final List<TestEntry> testEntries)
-    {
-        for (final TestEntry testEntry : testEntries)
+        if (hasPartiallyProcessedFiles)
+        {
+            Files.createFile(new File(tempDirA, RecordingLog.RECORDING_LOG_MIGRATED_FILE_NAME).toPath());
+            Files.createFile(new File(tempDirA, RecordingLog.RECORDING_LOG_NEW_FILE_NAME).toPath());
+        }
+
+        final RecordingLog recordingLogMigrated = new RecordingLog(tempDirA, false);
+        final RecordingLog recordingLogNew = new RecordingLog(tempDirB, true);
+
+        for (final TestEntry testEntry : generateData())
         {
-            recordingLog.append(
+            recordingLogNew.append(
                 testEntry.entryType,
                 testEntry.recordingId,
                 testEntry.leadershipTermId,
@@ -180,28 +193,6 @@ private static void appendToRecordingLog(final RecordingLog recordingLog, final
                 testEntry.serviceId,
                 testEntry.endpoint);
         }
-    }
-
-    @Test
-    void verifyDataRemainsConsistent()
-    {
-        assertEquals(generateData(), generateData());
-    }
-
-    @Test
-    void shouldLoadOldVersionAndMigrate(@TempDir final File tempDirA, @TempDir final File tempDirB) throws IOException
-    {
-        assertNotEquals(tempDirA, tempDirB);
-
-        final File parentDir = new File("src/test/resources/v1_45_x");
-        final File oldFile = new File(parentDir, RecordingLog.RECORDING_LOG_FILE_NAME);
-        final File tempOldFile = new File(tempDirA, RecordingLog.RECORDING_LOG_FILE_NAME);
-
-        Files.copy(oldFile.toPath(), tempOldFile.toPath());
-
-        final RecordingLog recordingLogMigrated = new RecordingLog(tempDirA, false);
-        final RecordingLog recordingLogNew = new RecordingLog(tempDirB, true);
-        appendToRecordingLog(recordingLogNew, generateData());
 
         assertEquals(recordingLogMigrated.entries(), recordingLogNew.entries());
     }
@@ -213,16 +204,16 @@ private int recordingType(final Random r)
         switch (type)
         {
             case 0: return ENTRY_TYPE_TERM;
-            case 1: return RecordingLog.ENTRY_TYPE_SNAPSHOT;
-            case 2: return RecordingLog.ENTRY_TYPE_STANDBY_SNAPSHOT;
+            case 1: return ENTRY_TYPE_SNAPSHOT;
+            case 2: return ENTRY_TYPE_STANDBY_SNAPSHOT;
         }
 
         throw new IllegalStateException();
     }
 
     private String endpoint(final Random r, final int type, final byte[] bs)
     {
-        if (RecordingLog.ENTRY_TYPE_STANDBY_SNAPSHOT == type)
+        if (ENTRY_TYPE_STANDBY_SNAPSHOT == type)
         {
             final int length = 50 + r.nextInt(50);
             for (int i = 0; i < length; i++)
diff --git a/aeron-cluster/src/test/java/io/aeron/cluster/UnversionedRecordingLog.java b/aeron-cluster/src/test/java/io/aeron/cluster/UnversionedRecordingLog.java
diff --git a/aeron-system-tests/src/test/java/io/aeron/cluster/ClusterToolTest.java b/aeron-system-tests/src/test/java/io/aeron/cluster/ClusterToolTest.java
@@ -22,6 +22,8 @@
 import io.aeron.test.SystemTestWatcher;
 import io.aeron.test.cluster.TestCluster;
 import io.aeron.test.cluster.TestNode;
+import org.agrona.MutableDirectBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -38,6 +40,7 @@
 import java.util.regex.Pattern;
 
 import static io.aeron.test.cluster.TestCluster.aCluster;
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
 import static java.nio.charset.StandardCharsets.US_ASCII;
 import static java.nio.file.StandardOpenOption.CREATE_NEW;
 import static java.nio.file.StandardOpenOption.WRITE;
@@ -220,7 +223,12 @@ void sortRecordingLogIsANoOpIfRecordLogIsEmpty(final @TempDir Path emptyClusterD
         final boolean result = ClusterTool.sortRecordingLog(clusterDir);
 
         assertFalse(result);
-        assertArrayEquals(new byte[0], Files.readAllBytes(logFile));
+
+        final MutableDirectBuffer header = new UnsafeBuffer(new byte[RecordingLog.HEADER_SIZE]);
+        header.putLong(0, RecordingLog.MAGIC_NUMBER, LITTLE_ENDIAN);
+        header.putInt(8, RecordingLog.SEMANTIC_VERSION, LITTLE_ENDIAN);
+
+        assertArrayEquals(header.byteArray(), Files.readAllBytes(logFile));
     }
 
     @Test

Original file line number	Diff line number	Diff line change
`@@ -670,6 +670,7 @@ void entriesInTheRecordingLogShouldBeSorted()`
`670`	`670`
`671`	`671`	`recordingLog.reload();`
`672`	`672`
	`673`	`+ assertEquals(sortedList.size(), recordingLog.entries().size());`
`673`	`674`	`assertEquals(sortedList, recordingLog.entries()); // reload from disc and re-sort`
`674`	`675`	`}`
`675`	`676`	`}`