@Test public void verifyCacheSizeMetricIsReported() throws IOException, TException { final String remotePathWithScheme = "file: final int readOffset = 0; final int readLength = 100; assertEquals(metrics.getGauges().get(BookKeeperMetrics.CacheMetric.CACHE_SIZE_GAUGE.getMetricName()).getValue(), 0); DataGen.populateFile(TEST_REMOTE_PATH); ReadResponse response = bookKeeper.readData(remotePathWithScheme, readOffset, readLength, TEST_FILE_LENGTH, TEST_LAST_MODIFIED, ClusterType.TEST_CLUSTER_MANAGER.ordinal()); final long mdSize = FileUtils.sizeOf(new File(CacheUtil.getMetadataFilePath(TEST_REMOTE_PATH, conf, response.getGenerationNumber()))); final int totalCacheSize = (int) BYTES.toMB(readLength + mdSize); assertEquals(metrics.getGauges().get(BookKeeperMetrics.CacheMetric.CACHE_SIZE_GAUGE.getMetricName()).getValue(), totalCacheSize); }
@Override public ReadResponse readData(String remotePath, long offset, int length, long fileSize, long lastModified, int clusterType) throws TException { if (CacheConfig.isParallelWarmupEnabled(conf)) { startRemoteFetchProcessor(); log.debug("Adding to the queue Path : " + remotePath + " Offset : " + offset + " Length " + length); fetchProcessor.addToProcessQueue(remotePath, offset, length, fileSize, lastModified); return new ReadResponse(false, UNKONWN_GENERATION_NUMBER); } else { return readDataInternal(remotePath, offset, length, fileSize, lastModified, clusterType); } }
BookKeeper implements BookKeeperService.Iface { @Override public ReadResponse readData(String remotePath, long offset, int length, long fileSize, long lastModified, int clusterType) throws TException { if (CacheConfig.isParallelWarmupEnabled(conf)) { startRemoteFetchProcessor(); log.debug("Adding to the queue Path : " + remotePath + " Offset : " + offset + " Length " + length); fetchProcessor.addToProcessQueue(remotePath, offset, length, fileSize, lastModified); return new ReadResponse(false, UNKONWN_GENERATION_NUMBER); } else { return readDataInternal(remotePath, offset, length, fileSize, lastModified, clusterType); } } }
BookKeeper implements BookKeeperService.Iface { @Override public ReadResponse readData(String remotePath, long offset, int length, long fileSize, long lastModified, int clusterType) throws TException { if (CacheConfig.isParallelWarmupEnabled(conf)) { startRemoteFetchProcessor(); log.debug("Adding to the queue Path : " + remotePath + " Offset : " + offset + " Length " + length); fetchProcessor.addToProcessQueue(remotePath, offset, length, fileSize, lastModified); return new ReadResponse(false, UNKONWN_GENERATION_NUMBER); } else { return readDataInternal(remotePath, offset, length, fileSize, lastModified, clusterType); } } BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics); @VisibleForTesting  BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics, Ticker ticker); }
BookKeeper implements BookKeeperService.Iface { @Override public ReadResponse readData(String remotePath, long offset, int length, long fileSize, long lastModified, int clusterType) throws TException { if (CacheConfig.isParallelWarmupEnabled(conf)) { startRemoteFetchProcessor(); log.debug("Adding to the queue Path : " + remotePath + " Offset : " + offset + " Length " + length); fetchProcessor.addToProcessQueue(remotePath, offset, length, fileSize, lastModified); return new ReadResponse(false, UNKONWN_GENERATION_NUMBER); } else { return readDataInternal(remotePath, offset, length, fileSize, lastModified, clusterType); } } BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics); @VisibleForTesting  BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics, Ticker ticker); @Override boolean isBookKeeperAlive(); @Override CacheStatusResponse getCacheStatus(CacheStatusRequest request); boolean isInitialized(); @VisibleForTesting void initializeClusterManager(int clusterType); @VisibleForTesting ClusterManager getClusterManagerInstance(ClusterType clusterType, Configuration config); @Override void setAllCached(String remotePath, long fileLength, long lastModified, long startBlock, long endBlock, int generationNumber); @VisibleForTesting long getTotalCacheWeight(); @Override Map<String, Double> getCacheMetrics(); @Override Map<String, Long> getReadRequestChainStats(); @Override ReadResponse readData(String remotePath, long offset, int length, long fileSize, long lastModified, int clusterType); @Override FileInfo getFileInfo(String remotePath); @VisibleForTesting FileMetadata getFileMetadata(String key); @Override void invalidateFileMetadata(String key); }
BookKeeper implements BookKeeperService.Iface { @Override public ReadResponse readData(String remotePath, long offset, int length, long fileSize, long lastModified, int clusterType) throws TException { if (CacheConfig.isParallelWarmupEnabled(conf)) { startRemoteFetchProcessor(); log.debug("Adding to the queue Path : " + remotePath + " Offset : " + offset + " Length " + length); fetchProcessor.addToProcessQueue(remotePath, offset, length, fileSize, lastModified); return new ReadResponse(false, UNKONWN_GENERATION_NUMBER); } else { return readDataInternal(remotePath, offset, length, fileSize, lastModified, clusterType); } } BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics); @VisibleForTesting  BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics, Ticker ticker); @Override boolean isBookKeeperAlive(); @Override CacheStatusResponse getCacheStatus(CacheStatusRequest request); boolean isInitialized(); @VisibleForTesting void initializeClusterManager(int clusterType); @VisibleForTesting ClusterManager getClusterManagerInstance(ClusterType clusterType, Configuration config); @Override void setAllCached(String remotePath, long fileLength, long lastModified, long startBlock, long endBlock, int generationNumber); @VisibleForTesting long getTotalCacheWeight(); @Override Map<String, Double> getCacheMetrics(); @Override Map<String, Long> getReadRequestChainStats(); @Override ReadResponse readData(String remotePath, long offset, int length, long fileSize, long lastModified, int clusterType); @Override FileInfo getFileInfo(String remotePath); @VisibleForTesting FileMetadata getFileMetadata(String key); @Override void invalidateFileMetadata(String key);  }