Make snapshot offset long to prevent overflow

The current implementation uses integer which can
easily overflow if the Snapshot grows over 2.14GB

JIRA: CONTROLLER-2086
Change-Id: Ibbe3d3e1667cf59137e057b31141033a826142e4
Signed-off-by: tibor.kral <tibor.kral@pantheon.tech>
Signed-off-by: Samuel Schneider <samuel.schneider@pantheon.tech>
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderInstallSnapshotState.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderInstallSnapshotState.java
index 68e82dc9a1151a05f47521188fb8304472c15c9b..a2617dc63960c676ccb7227a7922361f76568323 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderInstallSnapshotState.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderInstallSnapshotState.java
@@ -38,9 +38,9 @@ public final class LeaderInstallSnapshotState implements AutoCloseable {
     private final int snapshotChunkSize;
     private final String logName;
     private ByteSource snapshotBytes;
-    private int offset = INITIAL_OFFSET;
+    private long offset = INITIAL_OFFSET;
     // the next snapshot chunk is sent only if the replyReceivedForOffset matches offset
-    private int replyReceivedForOffset = -1;
+    private long replyReceivedForOffset = -1;
     // if replyStatus is false, the previous chunk is attempted
     private boolean replyStatus = false;
     private int chunkIndex = FIRST_CHUNK_INDEX;
@@ -75,8 +75,8 @@ public final class LeaderInstallSnapshotState implements AutoCloseable {
         chunkIndex = FIRST_CHUNK_INDEX;
     }
 
-    int incrementOffset() {
-        // if offset is -1 doesnt matter whether it was the initial value or reset, move the offset to 0 to begin with
+    private long incrementOffset() {
+        // if offset is -1 doesn't matter whether it was the initial value or reset, move the offset to 0 to begin with
         if (offset == INITIAL_OFFSET) {
             offset = 0;
         } else {
@@ -139,7 +139,7 @@ public final class LeaderInstallSnapshotState implements AutoCloseable {
     byte[] getNextChunk() throws IOException {
         // increment offset to indicate next chunk is in flight, canSendNextChunk() wont let us hit this again until,
         // markSendStatus() is called with either success or failure
-        int start = incrementOffset();
+        final var start = incrementOffset();
         if (replyStatus || currentChunk == null) {
             int size = snapshotChunkSize;
             if (snapshotChunkSize > snapshotSize) {
@@ -149,11 +149,11 @@ public final class LeaderInstallSnapshotState implements AutoCloseable {
             }
 
             currentChunk = new byte[size];
-            int numRead = snapshotInputStream.read(currentChunk);
+            final var numRead = snapshotInputStream.read(currentChunk);
             if (numRead != size) {
                 throw new IOException(String.format(
-                        "The # of bytes read from the input stream, %d,"
-                                + "does not match the expected # %d", numRead, size));
+                        "The # of bytes read from the input stream, %d, does not match the expected # %d",
+                        numRead, size));
             }
 
             nextChunkHashCode = Arrays.hashCode(currentChunk);

diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderInstallSnapshotStateTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderInstallSnapshotStateTest.java
new file mode 100644 (file)
index 0000000..aa07181
--- /dev/null
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderInstallSnapshotStateTest.java
@@ -0,0 +1,100 @@
+/*
+ * Copyright (c) 2023 PANTHEON.tech, s.r.o. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.raft.behaviors;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.common.io.ByteSource;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+import java.util.Objects;
+import org.junit.Test;
+
+public class LeaderInstallSnapshotStateTest {
+    // Prime number on purpose
+    private static final int CHUNK_SIZE = 9_999_991;
+    // More than Integer.MAX_VALUE
+    private static final long SIZE = 4_294_967_294L;
+
+    @Test
+    public void testSnapshotLongerThanInteger() throws IOException {
+        try (var fts = new LeaderInstallSnapshotState(CHUNK_SIZE, "test")) {
+            fts.setSnapshotBytes(new MockByteSource(SIZE));
+
+            int chunkIndex = 0;
+            long offset = 0;
+            long expectedChunkSize = CHUNK_SIZE;
+            while (offset < SIZE) {
+                offset = offset + CHUNK_SIZE;
+                if (offset > SIZE) {
+                    // We reached last chunk
+                    expectedChunkSize = CHUNK_SIZE - (offset - SIZE);
+                    offset = SIZE;
+                }
+                chunkIndex ++;
+                final byte[] chunk = fts.getNextChunk();
+                assertEquals("byte size not matching for chunk:", expectedChunkSize, chunk.length);
+                assertEquals("chunk index not matching", chunkIndex, fts.getChunkIndex());
+                fts.markSendStatus(true);
+                if (!fts.isLastChunk(chunkIndex)) {
+                    fts.incrementChunkIndex();
+                }
+            }
+
+            assertEquals("totalChunks not matching", chunkIndex, fts.getTotalChunks());
+        }
+    }
+
+    private static final class MockByteSource extends ByteSource {
+        private final long size;
+
+        private MockByteSource(final long size) {
+            this.size = size;
+        }
+
+        @Override
+        public long size() {
+            return size;
+        }
+
+        @Override
+        public InputStream openStream() {
+            return new MockInputStream(size);
+        }
+    }
+
+    private static final class MockInputStream extends InputStream {
+        private long remaining;
+
+        MockInputStream(final long size) {
+            remaining = size;
+        }
+
+        @Override
+        public int read() {
+            if (remaining > 0) {
+                remaining--;
+                return 0;
+            }
+            return -1;
+        }
+
+        @Override
+        public int read(final byte[] bytes, final int off, final int len) {
+            Objects.checkFromIndexSize(off, len, bytes.length);
+            if (remaining <= 0) {
+                return -1;
+            }
+            final int count = len <= remaining ? len : (int) remaining;
+            Arrays.fill(bytes, off, off + count, (byte) 0);
+            remaining -= count;
+            return count;
+        }
+    }
+}