opendaylight/md-sal/sal-akka-segmented-journal/src/test/java/org/opendaylight/controller/akka/segjournal/SegmentedFileJournalTest.java

   1 /*
   2  * Copyright (c) 2019 Pantheon Technologies, s.r.o. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.akka.segjournal;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertFalse;
  12 import static org.junit.Assert.assertNull;
  13 import static org.junit.Assert.assertTrue;
  14 import static org.mockito.ArgumentMatchers.any;
  15 import static org.mockito.Mockito.doNothing;
  16 import static org.mockito.Mockito.reset;
  17 import static org.mockito.Mockito.times;
  18 import static org.mockito.Mockito.verify;
  19
  20 import akka.actor.ActorRef;
  21 import akka.actor.ActorSystem;
  22 import akka.actor.PoisonPill;
  23 import akka.persistence.AtomicWrite;
  24 import akka.persistence.PersistentRepr;
  25 import akka.testkit.CallingThreadDispatcher;
  26 import akka.testkit.javadsl.TestKit;
  27 import io.atomix.storage.StorageLevel;
  28 import java.io.File;
  29 import java.io.IOException;
  30 import java.io.Serializable;
  31 import java.nio.file.Files;
  32 import java.nio.file.Path;
  33 import java.util.ArrayList;
  34 import java.util.List;
  35 import java.util.Optional;
  36 import java.util.function.Consumer;
  37 import java.util.stream.Collectors;
  38 import org.apache.commons.io.FileUtils;
  39 import org.junit.After;
  40 import org.junit.AfterClass;
  41 import org.junit.Before;
  42 import org.junit.BeforeClass;
  43 import org.junit.Test;
  44 import org.junit.runner.RunWith;
  45 import org.mockito.Mock;
  46 import org.mockito.junit.MockitoJUnitRunner;
  47 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.AsyncMessage;
  48 import org.opendaylight.controller.akka.segjournal.SegmentedJournalActor.WriteMessages;
  49 import scala.concurrent.Future;
  50
  51 @RunWith(MockitoJUnitRunner.StrictStubs.class)
  52 public class SegmentedFileJournalTest {
  53     private static final File DIRECTORY = new File("target/sfj-test");
  54     private static final int SEGMENT_SIZE = 1024 * 1024;
  55     private static final int MESSAGE_SIZE = 512 * 1024;
  56
  57     private static ActorSystem SYSTEM;
  58
  59     @Mock
  60     private Consumer<PersistentRepr> firstCallback;
  61
  62     private TestKit kit;
  63     private ActorRef actor;
  64
  65     @BeforeClass
  66     public static void beforeClass() {
  67         SYSTEM = ActorSystem.create("test");
  68     }
  69
  70     @AfterClass
  71     public static void afterClass() {
  72         TestKit.shutdownActorSystem(SYSTEM);
  73         SYSTEM = null;
  74     }
  75
  76     @Before
  77     public void before() {
  78         kit = new TestKit(SYSTEM);
  79         FileUtils.deleteQuietly(DIRECTORY);
  80         actor = actor();
  81     }
  82
  83     @After
  84     public void after() {
  85         actor.tell(PoisonPill.getInstance(), ActorRef.noSender());
  86     }
  87
  88     @Test
  89     public void testDeleteAfterStop() {
  90         // Preliminary setup
  91         final WriteMessages write = new WriteMessages();
  92         final Future<Optional<Exception>> first = write.add(AtomicWrite.apply(PersistentRepr.apply("first", 1, "foo",
  93             null, false, kit.getRef(), "uuid")));
  94         final Future<Optional<Exception>> second = write.add(AtomicWrite.apply(PersistentRepr.apply("second", 2, "foo",
  95             null, false, kit.getRef(), "uuid")));
  96         actor.tell(write, ActorRef.noSender());
  97         assertFalse(getFuture(first).isPresent());
  98         assertFalse(getFuture(second).isPresent());
  99
 100         assertHighestSequenceNr(2);
 101         assertReplayCount(2);
 102
 103         deleteEntries(1);
 104
 105         assertHighestSequenceNr(2);
 106         assertReplayCount(1);
 107
 108         // Restart actor
 109         actor.tell(PoisonPill.getInstance(), ActorRef.noSender());
 110         actor = actor();
 111
 112         // Check if state is retained
 113         assertHighestSequenceNr(2);
 114         assertReplayCount(1);
 115     }
 116
 117     @Test
 118     public void testSegmentation() throws IOException {
 119         // We want to have roughly three segments
 120         final LargePayload payload = new LargePayload();
 121
 122         final WriteMessages write = new WriteMessages();
 123         final List<Future<Optional<Exception>>> requests = new ArrayList<>();
 124
 125         // Each payload is half of segment size, plus some overhead, should result in two segments being present
 126         for (int i = 1; i <= SEGMENT_SIZE * 3 / MESSAGE_SIZE; ++i) {
 127             requests.add(write.add(AtomicWrite.apply(PersistentRepr.apply(payload, i, "foo", null, false, kit.getRef(),
 128                 "uuid"))));
 129         }
 130
 131         actor.tell(write, ActorRef.noSender());
 132         requests.forEach(future -> assertFalse(getFuture(future).isPresent()));
 133
 134         assertFileCount(2, 1);
 135
 136         // Delete all but the last entry
 137         deleteEntries(requests.size());
 138
 139         assertFileCount(1, 1);
 140     }
 141
 142     @Test
 143     public void testComplexDeletesAndPartialReplays() throws Exception {
 144         for (int i = 0; i <= 4; i++) {
 145             writeBigPaylod();
 146         }
 147
 148         assertFileCount(10, 1);
 149
 150         // delete including index 3, so get rid of the first segment
 151         deleteEntries(3);
 152         assertFileCount(9, 1);
 153
 154         // get rid of segments 2(index 4-6) and 3(index 7-9)
 155         deleteEntries(9);
 156         assertFileCount(7, 1);
 157
 158         // get rid of all segments except the last one
 159         deleteEntries(27);
 160         assertFileCount(1, 1);
 161
 162         restartActor();
 163
 164         // Check if state is retained
 165         assertHighestSequenceNr(30);
 166         // 28,29,30 replayed
 167         assertReplayCount(3);
 168
 169
 170         deleteEntries(28);
 171         restartActor();
 172
 173         assertHighestSequenceNr(30);
 174         // 29,30 replayed
 175         assertReplayCount(2);
 176
 177         deleteEntries(29);
 178         restartActor();
 179
 180         // 30 replayed
 181         assertReplayCount(1);
 182
 183         deleteEntries(30);
 184         restartActor();
 185
 186         // nothing replayed
 187         assertReplayCount(0);
 188     }
 189
 190     private void restartActor() {
 191         actor.tell(PoisonPill.getInstance(), ActorRef.noSender());
 192         actor = actor();
 193     }
 194
 195     private void writeBigPaylod() {
 196         final LargePayload payload = new LargePayload();
 197
 198         final WriteMessages write = new WriteMessages();
 199         final List<Future<Optional<Exception>>> requests = new ArrayList<>();
 200
 201         // Each payload is half of segment size, plus some overhead, should result in two segments being present
 202         for (int i = 1; i <= SEGMENT_SIZE * 3 / MESSAGE_SIZE; ++i) {
 203             requests.add(write.add(AtomicWrite.apply(PersistentRepr.apply(payload, i, "foo", null, false, kit.getRef(),
 204                     "uuid"))));
 205         }
 206
 207         actor.tell(write, ActorRef.noSender());
 208         requests.forEach(future -> assertFalse(getFuture(future).isPresent()));
 209     }
 210
 211     private ActorRef actor() {
 212         return kit.childActorOf(SegmentedJournalActor.props("foo", DIRECTORY, StorageLevel.DISK, MESSAGE_SIZE,
 213             SEGMENT_SIZE).withDispatcher(CallingThreadDispatcher.Id()));
 214     }
 215
 216     private void deleteEntries(final long deleteTo) {
 217         final AsyncMessage<Void> delete = SegmentedJournalActor.deleteMessagesTo(deleteTo);
 218         actor.tell(delete, ActorRef.noSender());
 219         assertNull(get(delete));
 220     }
 221
 222     private void assertHighestSequenceNr(final long expected) {
 223         AsyncMessage<Long> highest = SegmentedJournalActor.readHighestSequenceNr(0);
 224         actor.tell(highest, ActorRef.noSender());
 225         assertEquals(expected, (long) get(highest));
 226     }
 227
 228     private void assertReplayCount(final int expected) {
 229         // Cast fixes an Eclipse warning 'generic array created'
 230         reset((Object) firstCallback);
 231         doNothing().when(firstCallback).accept(any(PersistentRepr.class));
 232         AsyncMessage<Void> replay = SegmentedJournalActor.replayMessages(0, Long.MAX_VALUE, Long.MAX_VALUE,
 233             firstCallback);
 234         actor.tell(replay, ActorRef.noSender());
 235         assertNull(get(replay));
 236         verify(firstCallback, times(expected)).accept(any(PersistentRepr.class));
 237     }
 238
 239     private static void assertFileCount(final long dataFiles, final long deleteFiles) throws IOException {
 240         List<File> contents = Files.list(DIRECTORY.toPath()).map(Path::toFile).collect(Collectors.toList());
 241         assertEquals(dataFiles, contents.stream().filter(file -> file.getName().startsWith("data-")).count());
 242         assertEquals(deleteFiles, contents.stream().filter(file -> file.getName().startsWith("delete-")).count());
 243     }
 244
 245     private static <T> T get(final AsyncMessage<T> message) {
 246         return getFuture(message.promise.future());
 247     }
 248
 249     private static <T> T getFuture(final Future<T> future) {
 250         assertTrue(future.isCompleted());
 251         return future.value().get().get();
 252     }
 253
 254     private static final class LargePayload implements Serializable {
 255         private static final long serialVersionUID = 1L;
 256
 257         final byte[] bytes = new byte[MESSAGE_SIZE / 2];
 258
 259     }
 260 }