ceph/src/rocksdb/java/src/test/java/org/rocksdb/WalFilterTest.java

   1 // Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5
   6 package org.rocksdb;
   7
   8 import org.junit.ClassRule;
   9 import org.junit.Rule;
  10 import org.junit.Test;
  11 import org.junit.rules.TemporaryFolder;
  12
  13 import java.util.ArrayList;
  14 import java.util.Arrays;
  15 import java.util.List;
  16 import java.util.Map;
  17
  18 import static org.assertj.core.api.Assertions.assertThat;
  19 import static org.rocksdb.util.TestUtil.*;
  20
  21 public class WalFilterTest {
  22
  23   @ClassRule
  24   public static final RocksMemoryResource rocksMemoryResource =
  25       new RocksMemoryResource();
  26
  27   @Rule
  28   public TemporaryFolder dbFolder = new TemporaryFolder();
  29
  30   @Test
  31   public void walFilter() throws RocksDBException {
  32     // Create 3 batches with two keys each
  33     final byte[][][] batchKeys = {
  34         new byte[][] {
  35             u("key1"),
  36             u("key2")
  37         },
  38         new byte[][] {
  39             u("key3"),
  40             u("key4")
  41         },
  42         new byte[][] {
  43             u("key5"),
  44             u("key6")
  45         }
  46
  47     };
  48
  49     final List<ColumnFamilyDescriptor> cfDescriptors = Arrays.asList(
  50         new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY),
  51         new ColumnFamilyDescriptor(u("pikachu"))
  52     );
  53     final List<ColumnFamilyHandle> cfHandles = new ArrayList<>();
  54
  55     // Test with all WAL processing options
  56     for (final WalProcessingOption option : WalProcessingOption.values()) {
  57       try (final Options options = optionsForLogIterTest();
  58            final DBOptions dbOptions = new DBOptions(options)
  59                .setCreateMissingColumnFamilies(true);
  60            final RocksDB db = RocksDB.open(dbOptions,
  61                dbFolder.getRoot().getAbsolutePath(),
  62                 cfDescriptors, cfHandles)) {
  63         try (final WriteOptions writeOptions = new WriteOptions()) {
  64           // Write given keys in given batches
  65           for (int i = 0; i < batchKeys.length; i++) {
  66             final WriteBatch batch = new WriteBatch();
  67             for (int j = 0; j < batchKeys[i].length; j++) {
  68               batch.put(cfHandles.get(0), batchKeys[i][j], dummyString(1024));
  69             }
  70             db.write(writeOptions, batch);
  71           }
  72         } finally {
  73           for (final ColumnFamilyHandle cfHandle : cfHandles) {
  74             cfHandle.close();
  75           }
  76           cfHandles.clear();
  77         }
  78       }
  79
  80       // Create a test filter that would apply wal_processing_option at the first
  81       // record
  82       final int applyOptionForRecordIndex = 1;
  83       try (final TestableWalFilter walFilter =
  84                new TestableWalFilter(option, applyOptionForRecordIndex)) {
  85
  86         try (final Options options = optionsForLogIterTest();
  87              final DBOptions dbOptions = new DBOptions(options)
  88                 .setWalFilter(walFilter)) {
  89
  90           try (final RocksDB db = RocksDB.open(dbOptions,
  91               dbFolder.getRoot().getAbsolutePath(),
  92               cfDescriptors, cfHandles)) {
  93
  94             try {
  95               assertThat(walFilter.logNumbers).isNotEmpty();
  96               assertThat(walFilter.logFileNames).isNotEmpty();
  97             } finally {
  98               for (final ColumnFamilyHandle cfHandle : cfHandles) {
  99                 cfHandle.close();
 100               }
 101               cfHandles.clear();
 102             }
 103           } catch (final RocksDBException e) {
 104             if (option != WalProcessingOption.CORRUPTED_RECORD) {
 105               // exception is expected when CORRUPTED_RECORD!
 106               throw e;
 107             }
 108           }
 109         }
 110       }
 111     }
 112   }
 113
 114
 115   private static class TestableWalFilter extends AbstractWalFilter {
 116     private final WalProcessingOption walProcessingOption;
 117     private final int applyOptionForRecordIndex;
 118     Map<Integer, Long> cfLognumber;
 119     Map<String, Integer> cfNameId;
 120     final List<Long> logNumbers = new ArrayList<>();
 121     final List<String> logFileNames = new ArrayList<>();
 122     private int currentRecordIndex = 0;
 123
 124     public TestableWalFilter(final WalProcessingOption walProcessingOption,
 125         final int applyOptionForRecordIndex) {
 126       super();
 127       this.walProcessingOption = walProcessingOption;
 128       this.applyOptionForRecordIndex = applyOptionForRecordIndex;
 129     }
 130
 131     @Override
 132     public void columnFamilyLogNumberMap(final Map<Integer, Long> cfLognumber,
 133         final Map<String, Integer> cfNameId) {
 134       this.cfLognumber = cfLognumber;
 135       this.cfNameId = cfNameId;
 136     }
 137
 138     @Override
 139     public LogRecordFoundResult logRecordFound(
 140         final long logNumber, final String logFileName, final WriteBatch batch,
 141         final WriteBatch newBatch) {
 142
 143       logNumbers.add(logNumber);
 144       logFileNames.add(logFileName);
 145
 146       final WalProcessingOption optionToReturn;
 147       if (currentRecordIndex == applyOptionForRecordIndex) {
 148         optionToReturn = walProcessingOption;
 149       }
 150       else {
 151         optionToReturn = WalProcessingOption.CONTINUE_PROCESSING;
 152       }
 153
 154       currentRecordIndex++;
 155
 156       return new LogRecordFoundResult(optionToReturn, false);
 157     }
 158
 159     @Override
 160     public String name() {
 161       return "testable-wal-filter";
 162     }
 163   }
 164 }