Changeset 4627


Ignore:
Timestamp:
2011-02-04 09:28:44 (2 years ago)
Author:
evert
Message:

BlobIncubatorMonitor? monitors the BlobIncubatorTable? for any unused (stale) blobs and removes them from the blobstore.
(#177)

Location:
trunk
Files:
3 added
24 edited
1 moved

Legend:

Unmodified
Added
Removed

  • trunk/cr/indexer/engine/src/test/java/org/lilyproject/indexer/engine/test/IndexerTest.java

    r4622 r4627  
    167167        RowLogShard walShard = new RowLogShardImpl("WS1", HBASE_PROXY.getConf(), wal, 100); 
    168168        wal.registerShard(walShard); 
    169         BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory); 
     169        BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory, false); 
    170170        repository = new HBaseRepository(typeManager, idGenerator, wal, HBASE_PROXY.getConf(), hbaseTableFactory, blobManager); 
    171171 

  • trunk/cr/linkindex/src/test/java/org/lilyproject/linkindex/test/LinkIndexTest.java

    r4621 r4627  
    9696        BlobStoreAccess dfsBlobStoreAccess = new DFSBlobStoreAccess(HBASE_PROXY.getBlobFS(), new Path("/lily/blobs")); 
    9797        SizeBasedBlobStoreAccessFactory blobStoreAccessFactory = new SizeBasedBlobStoreAccessFactory(dfsBlobStoreAccess); 
    98         BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory); 
     98        BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory, false); 
    9999 
    100100        rowLogConfMgr = new RowLogConfigurationManagerImpl(zk); 

  • trunk/cr/process/client/src/main/java/org/lilyproject/client/LilyClient.java

    r4621 r4627  
    191191        blobStoreAccessFactory.addBlobStoreAccess(200000, hbaseBlobStoreAccess); 
    192192         
    193         return new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory); 
     193        return new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory, true); 
    194194    } 
    195195     

  • trunk/cr/process/server/conf/general/tables.xml

    r4623 r4627  
    5353    </splits> 
    5454  </table> 
     55   
     56  <table name="blobincubator"> 
     57    <splits> 
     58      <regionCount>20</regionCount> 
     59      <splitKeys></splitKeys> 
     60    </splits> 
     61  </table> 
    5562 
    5663</tables> 

  • trunk/cr/process/server/quick-conf/general/tables.xml

    r4625 r4627  
    2121    </splits> 
    2222  </table> 
     23   
     24  <table name="blobincubator"> 
     25    <splits> 
     26      <regionCount>1</regionCount> 
     27    </splits> 
     28  </table> 
    2329 
    2430</tables> 

  • trunk/cr/process/server/repository-module/pom.xml

    r4607 r4627  
    5353      <artifactId>spring-beans</artifactId> 
    5454    </dependency> 
     55     
     56    <dependency> 
     57      <groupId>org.kauriproject</groupId> 
     58      <artifactId>kauri-conf</artifactId> 
     59      <version>${version.kauri}</version> 
     60    </dependency> 
    5561  </dependencies> 
    5662 

  • trunk/cr/process/server/repository-module/src/main/java/org/lilyproject/server/modules/repository/BlobManagerSetup.java

    r4621 r4627  
    3232import org.lilyproject.util.repo.DfsUri; 
    3333 
    34 public class BlobStoreConfig { 
     34public class BlobManagerSetup { 
    3535    static BlobManager get(URI dfsUri, Configuration configuration, HBaseTableFactory tableFactory) throws IOException { 
    3636        FileSystem fs = FileSystem.get(DfsUri.getBaseDfsUri(dfsUri), configuration); 
    … …  
    4343        blobStoreAccessFactory.addBlobStoreAccess(5000, inlineBlobStoreAccess); 
    4444        blobStoreAccessFactory.addBlobStoreAccess(200000, hbaseBlobStoreAccess); 
    45         return new BlobManagerImpl(tableFactory, blobStoreAccessFactory); 
     45        return new BlobManagerImpl(tableFactory, blobStoreAccessFactory, false); 
    4646    } 
    4747} 

  • trunk/cr/process/server/repository-module/src/main/kauri/spring/services.xml

    r4621 r4627  
    7575  <bean 
    7676    id="blobManager" 
    77     class="org.lilyproject.server.modules.repository.BlobStoreConfig" 
     77    class="org.lilyproject.server.modules.repository.BlobManagerSetup" 
    7878    factory-method="get"> 
    7979    <constructor-arg> 
    … …  
    8484    <constructor-arg ref="hbaseConf"/> 
    8585    <constructor-arg ref="hbaseTableFactory"/> 
     86  </bean> 
     87   
     88  <bean id="blobIncubatorMonitorSetup" class="org.lilyproject.server.modules.repository.BlobIncubatorMonitorSetup"> 
     89    <constructor-arg ref="zooKeeper"/> 
     90    <constructor-arg ref="hbaseTableFactory"/> 
     91    <constructor-arg ref="blobManager"/> 
     92    <constructor-arg ref="typeManager"/> 
     93    <constructor-arg><kauri:conf path="blobmanager"/></constructor-arg> 
    8694  </bean> 
    8795 

  • trunk/cr/repository-api-tutorial/src/test/java/org/lilyproject/repository/api/tutorial/TutorialTest.java

    r4621 r4627  
    104104        SizeBasedBlobStoreAccessFactory blobStoreAccessFactory = new SizeBasedBlobStoreAccessFactory(dfsBlobStoreAccess); 
    105105        blobStoreAccessFactory.addBlobStoreAccess(Long.MAX_VALUE, dfsBlobStoreAccess); 
    106         BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory); 
     106        BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory, false); 
    107107        setupWal(); 
    108108        repository = new HBaseRepository(typeManager, idGenerator, wal, configuration, hbaseTableFactory, blobManager); 

  • trunk/cr/repository/api/src/main/java/org/lilyproject/repository/api/BlobManager.java

    r4621 r4627  
    7878    void register(BlobStoreAccess blobStoreAccess); 
    7979 
     80    void delete(byte[] blobKey) throws BlobException; 
    8081} 

  • trunk/cr/repository/api/src/main/java/org/lilyproject/repository/api/Repository.java

    r4621 r4627  
    241241    OutputStream getOutputStream(Blob blob) throws BlobException, InterruptedException; 
    242242 
    243      
    244  
    245243    /** 
    246244     * Returns an {@link InputStream} from which the binary data of a blob can be read. 
    247      * The blob can only be retrieved by referring to location of the blob in the record  
     245     *  
     246     * <p>The blob can only be retrieved by referring to location of the blob in the record  
    248247     * and field where it is used.    
    249248     * @param recordId the id of the record containing the blob 

  • trunk/cr/repository/impl/src/main/java/org/lilyproject/repository/impl/BlobManagerImpl.java

    r4621 r4627  
    3434    private Log log = LogFactory.getLog(getClass()); 
    3535     
    36     private static final byte[] INCUBATE = new byte[]{(byte)-1}; 
     36    protected static final byte[] INCUBATE = new byte[]{(byte)-1}; 
    3737    private HTableInterface blobIncubatorTable; 
    3838 
    … …  
    4141    private BlobStoreAccessRegistry registry; 
    4242 
    43     public BlobManagerImpl(HBaseTableFactory hbaseTableFactory, BlobStoreAccessFactory blobStoreAccessFactory) throws IOException { 
     43    public BlobManagerImpl(HBaseTableFactory hbaseTableFactory, BlobStoreAccessFactory blobStoreAccessFactory, boolean clientMode) throws IOException { 
    4444        this.factory = blobStoreAccessFactory; 
    45         blobIncubatorTable = LilyHBaseSchema.getBlobIncubatorTable(hbaseTableFactory); 
     45        blobIncubatorTable = LilyHBaseSchema.getBlobIncubatorTable(hbaseTableFactory, clientMode); 
    4646        registry = new BlobStoreAccessRegistry(this); 
    4747        registry.setBlobStoreAccessFactory(blobStoreAccessFactory); 
    … …  
    164164        return blob; 
    165165    } 
     166 
     167    public void delete(byte[] blobKey) throws BlobException { 
     168        registry.delete(blobKey); 
     169         
     170    } 
    166171} 

  • trunk/cr/repository/impl/src/main/java/org/lilyproject/repository/impl/BlobStoreAccessRegistry.java

    r4621 r4627  
    8484    } 
    8585     
     86    public void delete(byte[] blobKey) throws BlobException { 
     87        Pair<String,byte[]> decodedKey = decode(blobKey); 
     88        BlobStoreAccess blobStoreAccess = registry.get(decodedKey.getV1()); 
     89        blobStoreAccess.delete(decodedKey.getV2()); 
     90    } 
     91     
    8692    static private byte[] encode(String id, byte[] blobKey) { 
    8793        byte[] bytes = new byte[0]; 

  • trunk/cr/repository/impl/src/main/java/org/lilyproject/repository/impl/ContainsValueComparator.java

    r4621 r4627  
    77import org.apache.hadoop.hbase.filter.WritableByteArrayComparable; 
    88import org.apache.hadoop.hbase.util.Bytes; 
    9 import org.lilyproject.repository.api.ValueType; 
    109 
    1110public class ContainsValueComparator extends WritableByteArrayComparable { 
    … …  
    4645    public int compareTo(byte[] theirValue) { 
    4746        int valueTypeCode = Bytes.toInt(valueTypeAndValue); 
    48         byte[] ourValue = Bytes.tail(valueTypeAndValue, valueTypeAndValue.length-Bytes.SIZEOF_INT); 
    49         if (theirValue == null && ourValue == null) 
     47        byte[] ourStoreKey = Bytes.tail(valueTypeAndValue, valueTypeAndValue.length-Bytes.SIZEOF_INT); 
     48        if (theirValue == null && ourStoreKey == null) 
    5049            return 0; 
    51         if (theirValue.length == 0 && ourValue.length == 0) 
     50        if (theirValue.length == 0 && ourStoreKey.length == 0) 
    5251            return 0; 
    53         if (theirValue.length < ourValue.length) 
     52        if (theirValue.length < ourStoreKey.length) 
    5453            return -1; 
    5554        if (theirValue[0] == (byte)(1)) { 
    … …  
    5756        } 
    5857        // Multivalue and hierarchical 
    59         int offset = 1; 
     58        int offset = 1; // First byte indicates if it was deleted or not 
    6059        if (2 == valueTypeCode) { 
    6160            int compareTo = -1; 
    6261            while (offset < theirValue.length) { 
    63                 int multivalueLength = Bytes.toInt(theirValue, offset); 
     62                int multivalueKeyLength = Bytes.toInt(theirValue, offset); // Length of the next multivalue key 
    6463                offset = offset + Bytes.SIZEOF_INT; 
    65                 int nextMultivalueOffset = offset+multivalueLength; 
    66                 while (offset < nextMultivalueOffset) { 
    67                     int valueLength = Bytes.toInt(theirValue, offset); 
     64                int stopIndex = offset + multivalueKeyLength; 
     65                while (offset < stopIndex) { 
     66                    int hierarchycalKeyLength = Bytes.toInt(theirValue, offset); // Length of the next hierarchy key 
    6867                    offset = offset + Bytes.SIZEOF_INT; 
    69                     compareTo = Bytes.compareTo(ourValue, 0, ourValue.length, theirValue, offset, valueLength); 
     68                    int valueLength = Bytes.toInt(theirValue, offset); // Length of the blob key 
     69                    // Don't increase offset here, it's calculated in the hierarchycalKeyLength 
     70                    compareTo = Bytes.compareTo(ourStoreKey, 0, ourStoreKey.length, theirValue, offset + Bytes.SIZEOF_INT, valueLength); 
    7071                    if (0 == compareTo) 
    7172                        return 0; 
    72                     offset = offset + valueLength; 
     73                    offset = offset + hierarchycalKeyLength; 
    7374                } 
    7475            } 
    … …  
    7980            int compareTo = -1; 
    8081            while (offset < theirValue.length) { 
    81                 int valueLength = Bytes.toInt(theirValue, offset); 
     82                int multiValueKeyLength = Bytes.toInt(theirValue, offset); // Length of the next multivalue or hierarchy key 
    8283                offset = offset + Bytes.SIZEOF_INT; 
    83                 compareTo = Bytes.compareTo(ourValue, 0, ourValue.length, theirValue, offset, valueLength); 
     84                int valueLength = Bytes.toInt(theirValue, offset); // Length of the blob key 
     85             // Don't increase offset here, it's calculated in the multivalueKeyLength 
     86                compareTo = Bytes.compareTo(ourStoreKey, 0, ourStoreKey.length, theirValue, offset + Bytes.SIZEOF_INT, valueLength); 
    8487                if (0 == compareTo) 
    8588                    return 0; 
    86                 offset = offset + valueLength; 
     89                offset = offset + multiValueKeyLength; 
    8790            } 
    8891            return -1; 
    8992        } 
    90         int valueLength = Bytes.toInt(theirValue, offset); 
    91         return Bytes.compareTo(ourValue, 0, ourValue.length, theirValue, offset, valueLength); 
     93        int valueLength = Bytes.toInt(theirValue, offset); // Length of the blob key, everything beyond that is mediatype, size, filename etc. 
     94        offset = offset + Bytes.SIZEOF_INT;  
     95        return Bytes.compareTo(ourStoreKey, 0, ourStoreKey.length, theirValue, offset, valueLength); 
    9296    } 
    9397} 

  • trunk/cr/repository/impl/src/main/java/org/lilyproject/repository/impl/FieldTypeImpl.java

    r4592 r4627  
    5252    } 
    5353     
    54     protected byte[] getIdBytes() { 
     54    public byte[] getIdBytes() { 
    5555        return idBytes; 
    5656    } 

  • trunk/cr/repository/impl/src/main/java/org/lilyproject/repository/impl/HBaseRepository.java

    r4621 r4627  
    12251225        for (BlobReference blobReference : blobs) { 
    12261226            FieldTypeImpl fieldType = (FieldTypeImpl)blobReference.getFieldType(); 
    1227             Get get = new Get(recordId.toBytes()); 
    1228             get.addColumn(columnFamily, fieldType.getIdBytes()); 
     1227            byte[] fieldTypeIdBytes = fieldType.getIdBytes(); 
     1228            byte[] recordIdBytes = recordId.toBytes(); 
    12291229            ValueType valueType = fieldType.getValueType(); 
     1230 
     1231            Get get = new Get(recordIdBytes); 
     1232            get.addColumn(columnFamily, fieldTypeIdBytes); 
    12301233            byte[] valueToCompare; 
    12311234            if (valueType.isMultiValue() && valueType.isHierarchical()) { 
    … …  
    12361239                valueToCompare = Bytes.toBytes(0); 
    12371240            } 
    1238             valueToCompare = Bytes.add(valueToCompare, valueType.getPrimitive().toBytes(blobReference.getBlob())); 
     1241            valueToCompare = Bytes.add(valueToCompare, blobReference.getBlob().getValue()); 
    12391242            WritableByteArrayComparable valueComparator = new ContainsValueComparator(valueToCompare); 
    12401243            Filter filter = new ValueFilter(CompareOp.EQUAL, valueComparator); 

  • trunk/cr/repository/impl/src/test/java/org/lilyproject/repository/impl/test/AbstractBlobStoreTest.java

    r4621 r4627  
    2323import java.io.OutputStream; 
    2424import java.net.URISyntaxException; 
    25 import java.util.Arrays; 
    26 import java.util.Random; 
     25import java.util.*; 
    2726 
    2827import org.apache.hadoop.conf.Configuration; 
    2928import org.apache.hadoop.fs.Path; 
     29import org.apache.hadoop.hbase.client.*; 
    3030import org.apache.hadoop.hbase.util.Bytes; 
    3131import org.junit.Test; 
    32 import org.lilyproject.repository.api.Blob; 
    33 import org.lilyproject.repository.api.BlobException; 
    34 import org.lilyproject.repository.api.BlobManager; 
    35 import org.lilyproject.repository.api.BlobNotFoundException; 
    36 import org.lilyproject.repository.api.BlobStoreAccess; 
    37 import org.lilyproject.repository.api.FieldType; 
    38 import org.lilyproject.repository.api.FieldTypeEntry; 
    39 import org.lilyproject.repository.api.FieldTypeNotFoundException; 
    40 import org.lilyproject.repository.api.HierarchyPath; 
    41 import org.lilyproject.repository.api.InvalidRecordException; 
    42 import org.lilyproject.repository.api.QName; 
    43 import org.lilyproject.repository.api.Record; 
    44 import org.lilyproject.repository.api.RecordException; 
    45 import org.lilyproject.repository.api.RecordId; 
    46 import org.lilyproject.repository.api.RecordNotFoundException; 
    47 import org.lilyproject.repository.api.RecordType; 
    48 import org.lilyproject.repository.api.RecordTypeNotFoundException; 
    49 import org.lilyproject.repository.api.Repository; 
    50 import org.lilyproject.repository.api.Scope; 
    51 import org.lilyproject.repository.api.TypeException; 
    52 import org.lilyproject.repository.api.TypeManager; 
    53 import org.lilyproject.repository.api.VersionNotFoundException; 
    54 import org.lilyproject.repository.impl.BlobManagerImpl; 
    55 import org.lilyproject.repository.impl.BlobStoreAccessRegistry; 
    56 import org.lilyproject.repository.impl.DFSBlobStoreAccess; 
    57 import org.lilyproject.repository.impl.HBaseBlobStoreAccess; 
    58 import org.lilyproject.repository.impl.InlineBlobStoreAccess; 
    59 import org.lilyproject.repository.impl.SizeBasedBlobStoreAccessFactory; 
     32import org.lilyproject.repository.api.*; 
     33import org.lilyproject.repository.impl.*; 
    6034import org.lilyproject.rowlog.api.RowLog; 
    6135import org.lilyproject.rowlog.api.RowLogConfig; 
    … …  
    8660    protected static Random random = new Random(); 
    8761    protected static BlobStoreAccessRegistry testBlobStoreAccessRegistry; 
     62    protected static BlobManager blobManager; 
    8863 
    8964    protected static void setupWal() throws Exception { 
    … …  
    10378        blobStoreAccessFactory.addBlobStoreAccess(50, inlineBlobStoreAccess); 
    10479        blobStoreAccessFactory.addBlobStoreAccess(1024, hbaseBlobStoreAccess); 
    105         return new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory); 
     80        return new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory, false); 
    10681    } 
    10782     
    … …  
    703678        } 
    704679    } 
    705  
     680     
     681    @Test 
     682    public void testBlobIncubatorMonitorUnusedBlob() throws Exception { 
     683        QName fieldName = new QName("test", "testBlobIncubatorMonitorUnusedBlob"); 
     684        FieldType fieldType = typeManager.newFieldType(typeManager.getValueType("BLOB", false, false), fieldName, 
     685                Scope.NON_VERSIONED); 
     686        fieldType = typeManager.createFieldType(fieldType); 
     687        RecordType recordType = typeManager.newRecordType(new QName(null, "testBlobIncubatorMonitorUnusedBlobRT")); 
     688        FieldTypeEntry fieldTypeEntry = typeManager.newFieldTypeEntry(fieldType.getId(), true); 
     689        recordType.addFieldTypeEntry(fieldTypeEntry); 
     690        recordType = typeManager.createRecordType(recordType); 
     691         
     692        // Incubate blob but never use it 
     693        byte[] bytes = new byte[3000];  
     694        random.nextBytes(bytes); 
     695        Blob blob = writeBlob(bytes, "aMediaType", "testCreate");   
     696         
     697        BlobIncubatorMonitor monitor = new BlobIncubatorMonitor(zooKeeper, hbaseTableFactory, blobManager, typeManager, 1000, 100); 
     698        monitor.startMonitoring(); 
     699        Thread.sleep(10000); 
     700        monitor.stopMonitoring(); 
     701         
     702        assertBlobDelete(true, blob); 
     703    } 
     704 
     705    @Test 
     706    public void testBlobIncubatorMonitorFailureAfterReservation() throws Exception { 
     707        QName fieldName = new QName("test", "testBlobIncubatorMonitorFailureAfterReservation"); 
     708        FieldType fieldType = typeManager.newFieldType(typeManager.getValueType("BLOB", false, false), fieldName, 
     709                Scope.NON_VERSIONED); 
     710        fieldType = typeManager.createFieldType(fieldType); 
     711        RecordType recordType = typeManager.newRecordType(new QName(null, "testBlobIncubatorMonitorFailureAfterReservationRT")); 
     712        FieldTypeEntry fieldTypeEntry = typeManager.newFieldTypeEntry(fieldType.getId(), true); 
     713        recordType.addFieldTypeEntry(fieldTypeEntry); 
     714        recordType = typeManager.createRecordType(recordType); 
     715         
     716    // This is the failure scenario where creating the record fails after reserving the blob 
     717        byte[] bytes = new byte[3000];  
     718        random.nextBytes(bytes); 
     719        Blob blob = writeBlob(bytes, "aMediaType", "testCreate"); 
     720        IdGeneratorImpl idGeneratorImpl = new IdGeneratorImpl(); 
     721        RecordId recordId = idGeneratorImpl.newRecordId(); 
     722        BlobReference blobReference = new BlobReference(blob, recordId, fieldType); 
     723        Set<BlobReference> blobs = new HashSet<BlobReference>(); 
     724        blobs.add(blobReference); 
     725        blobManager.reserveBlobs(blobs); 
     726         
     727        BlobIncubatorMonitor monitor = new BlobIncubatorMonitor(zooKeeper, hbaseTableFactory, blobManager, typeManager, 1000, 100); 
     728        monitor.startMonitoring(); 
     729        Thread.sleep(10000); 
     730        monitor.stopMonitoring(); 
     731      
     732        assertBlobDelete(true, blob); 
     733    } 
     734     
     735    @Test 
     736    public void testBlobIncubatorMonitorFailureBeforeRemovingReservation() throws Exception { 
     737        QName fieldName = new QName("test", "testBlobIncubatorMonitorFailureBeforeRemovingReservation"); 
     738        FieldType fieldType = typeManager.newFieldType(typeManager.getValueType("BLOB", false, false), fieldName, 
     739                Scope.NON_VERSIONED); 
     740        fieldType = typeManager.createFieldType(fieldType); 
     741        RecordType recordType = typeManager.newRecordType(new QName(null, "testBlobIncubatorMonitorFailureBeforeRemovingReservation")); 
     742        FieldTypeEntry fieldTypeEntry = typeManager.newFieldTypeEntry(fieldType.getId(), true); 
     743        recordType.addFieldTypeEntry(fieldTypeEntry); 
     744        recordType = typeManager.createRecordType(recordType); 
     745         
     746    // This is the failure scenario where creating the record fails after reserving the blob 
     747        byte[] bytes = new byte[3000];  
     748        random.nextBytes(bytes); 
     749        Blob blob = writeBlob(bytes, "aMediaType", "testCreate"); 
     750        IdGeneratorImpl idGeneratorImpl = new IdGeneratorImpl(); 
     751        RecordId recordId = idGeneratorImpl.newRecordId(); 
     752        BlobReference blobReference = new BlobReference(blob, recordId, fieldType); 
     753        Set<BlobReference> blobs = new HashSet<BlobReference>(); 
     754        blobs.add(blobReference); 
     755        repository.newRecord(); 
     756        Record record = repository.newRecord(); 
     757        record.setRecordType(recordType.getName()); 
     758        record.setField(fieldName, blob); 
     759        record = repository.create(record); 
     760         
     761        // Faking failure 
     762        HTableInterface blobIncubatorTable = LilyHBaseSchema.getBlobIncubatorTable(hbaseTableFactory, true); 
     763        Put put = new Put(blob.getValue()); 
     764        put.add(LilyHBaseSchema.BlobIncubatorCf.REF.bytes, LilyHBaseSchema.BlobIncubatorColumn.RECORD.bytes, record.getId().toBytes()); 
     765        put.add(LilyHBaseSchema.BlobIncubatorCf.REF.bytes, LilyHBaseSchema.BlobIncubatorColumn.FIELD.bytes, ((FieldTypeImpl)fieldType).getIdBytes()); 
     766        blobIncubatorTable.put(put); 
     767         
     768        BlobIncubatorMonitor monitor = new BlobIncubatorMonitor(zooKeeper, hbaseTableFactory, blobManager, typeManager, 1000, 100); 
     769        monitor.startMonitoring(); 
     770        Thread.sleep(10000); 
     771        monitor.stopMonitoring(); 
     772      
     773        assertBlobDelete(false, blob); 
     774        Get get = new Get(blob.getValue()); 
     775        Result result = blobIncubatorTable.get(get); 
     776        assertTrue(result == null || result.isEmpty()); 
     777    } 
     778         
    706779    private void assertBlobDelete(boolean expectDelete, Blob blob) throws BlobNotFoundException, BlobException { 
    707780        if (expectDelete) { 
    … …  
    715788        } 
    716789    } 
    717  
    718790     
    719791    @Test 

  • trunk/cr/repository/impl/src/test/java/org/lilyproject/repository/impl/test/AvroRepositoryTest.java

    r4621 r4627  
    6262        DFSBlobStoreAccess dfsBlobStoreAccess = new DFSBlobStoreAccess(HBASE_PROXY.getBlobFS(), new Path("/lily/blobs")); 
    6363        BlobStoreAccessFactory blobStoreAccessFactory = new SizeBasedBlobStoreAccessFactory(dfsBlobStoreAccess); 
    64         BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory); 
     64        BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory, false); 
    6565        setupWal(); 
    6666        serverRepository = new HBaseRepository(serverTypeManager, idGenerator, wal, configuration, hbaseTableFactory, blobManager); 

  • trunk/cr/repository/impl/src/test/java/org/lilyproject/repository/impl/test/AvroTypeManagerFieldTypeTest.java

    r4621 r4627  
    8686        DFSBlobStoreAccess dfsBlobStoreAccess = new DFSBlobStoreAccess(HBASE_PROXY.getBlobFS(), new Path("/lily/blobs")); 
    8787        BlobStoreAccessFactory blobStoreAccessFactory = new SizeBasedBlobStoreAccessFactory(dfsBlobStoreAccess); 
    88         BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory); 
     88        BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory, false); 
    8989        serverRepository = new HBaseRepository(serverTypeManager, idGenerator, wal, configuration, hbaseTableFactory, blobManager); 
    9090         

  • trunk/cr/repository/impl/src/test/java/org/lilyproject/repository/impl/test/AvroTypeManagerRecordTypeTest.java

    r4621 r4627  
    9797        DFSBlobStoreAccess dfsBlobStoreAccess = new DFSBlobStoreAccess(HBASE_PROXY.getBlobFS(), new Path("/lily/blobs")); 
    9898        BlobStoreAccessFactory blobStoreAccessFactory = new SizeBasedBlobStoreAccessFactory(dfsBlobStoreAccess); 
    99         BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory); 
     99        BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory, false); 
    100100        setupWal(); 
    101101        serverRepository = new HBaseRepository(serverTypeManager, idGenerator, wal, configuration, hbaseTableFactory, blobManager); 

  • trunk/cr/repository/impl/src/test/java/org/lilyproject/repository/impl/test/BlobStoreTest.java

    r4621 r4627  
    4444        typeManager = new HBaseTypeManager(idGenerator, configuration, zooKeeper, hbaseTableFactory); 
    4545        setupWal(); 
    46         BlobManager blobManager = setupBlobManager(); 
     46        blobManager = setupBlobManager(); 
    4747        repository = new HBaseRepository(typeManager, idGenerator, wal, configuration, hbaseTableFactory, blobManager); 
    4848        // Create a blobStoreAccessRegistry for testing purposes 

  • trunk/cr/repository/impl/src/test/java/org/lilyproject/repository/impl/test/HBaseRepositoryTest.java

    r4621 r4627  
    5353        DFSBlobStoreAccess dfsBlobStoreAccess = new DFSBlobStoreAccess(HBASE_PROXY.getBlobFS(), new Path("/lily/blobs")); 
    5454        blobStoreAccessFactory = new SizeBasedBlobStoreAccessFactory(dfsBlobStoreAccess); 
    55         BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory,blobStoreAccessFactory); 
     55        BlobManager blobManager = new BlobManagerImpl(hbaseTableFactory,blobStoreAccessFactory, false); 
    5656        setupWal(); 
    5757        repository = new HBaseRepository(typeManager, idGenerator, wal, configuration, hbaseTableFactory, blobManager); 

  • trunk/cr/repository/impl/src/test/java/org/lilyproject/repository/impl/test/RemoteBlobStoreTest.java

    r4621 r4627  
    7575        typeManager = new RemoteTypeManager(new InetSocketAddress(lilyServer.getPort()), 
    7676                remoteConverter, idGenerator, zooKeeper); 
    77         BlobManager blobManager = setupBlobManager(); 
     77        blobManager = setupBlobManager(); 
    7878        repository = new RemoteRepository(new InetSocketAddress(lilyServer.getPort()), remoteConverter, 
    7979                (RemoteTypeManager)typeManager, idGenerator, blobManager); 

  • trunk/cr/repository/impl/src/test/java/org/lilyproject/repository/impl/test/ValueTypeTest.java

    r4621 r4627  
    8787        InlineBlobStoreAccess inlineBlobStoreAccess = new InlineBlobStoreAccess(); 
    8888        BlobStoreAccessFactory blobStoreAccessFactory = new SizeBasedBlobStoreAccessFactory(inlineBlobStoreAccess); 
    89         repository = new HBaseRepository(typeManager, idGenerator, initializeWal(HBASE_PROXY.getConf()), HBASE_PROXY.getConf(), hbaseTableFactory, new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory)); 
     89        repository = new HBaseRepository(typeManager, idGenerator, initializeWal(HBASE_PROXY.getConf()), HBASE_PROXY.getConf(), hbaseTableFactory, new BlobManagerImpl(hbaseTableFactory, blobStoreAccessFactory, false)); 
    9090    } 
    9191 

  • trunk/global/hbase-util/src/main/java/org/lilyproject/util/hbase/LilyHBaseSchema.java

    r4621 r4627  
    5454    } 
    5555     
    56     public static HTableInterface getBlobIncubatorTable(HBaseTableFactory tableFactory) throws IOException { 
    57         return tableFactory.getTable(blobIncubatorDescriptor); 
     56    public static HTableInterface getBlobIncubatorTable(HBaseTableFactory tableFactory, boolean clientMode) throws IOException { 
     57        return tableFactory.getTable(blobIncubatorDescriptor, !clientMode); 
    5858    } 
    5959 
Note: See TracChangeset for help on using the changeset viewer.