JournaledEditsCache { void storeEdits(byte[] inputData, long newStartTxn, long newEndTxn, int newLayoutVersion) { if (newStartTxn < 0 || newEndTxn < newStartTxn) { Journal.LOG.error(String.format("Attempted to cache data of length %d " + "with newStartTxn %d and newEndTxn %d", inputData.length, newStartTxn, newEndTxn)); return; } try (AutoCloseableLock l = writeLock.acquire()) { if (newLayoutVersion != layoutVersion) { try { updateLayoutVersion(newLayoutVersion, newStartTxn); } catch (IOException ioe) { Journal.LOG.error(String.format("Unable to save new edits [%d, %d] " + "due to exception when updating to new layout version %d", newStartTxn, newEndTxn, newLayoutVersion), ioe); return; } } else if (lowestTxnId == INVALID_TXN_ID) { Journal.LOG.info("Initializing edits cache starting from txn ID " + newStartTxn); initialize(newStartTxn); } else if (highestTxnId + 1 != newStartTxn) { Journal.LOG.error(String.format("Edits cache is out of sync; " + "looked for next txn id at %d but got start txn id for " + "cache put request at %d. Reinitializing at new request.", highestTxnId + 1, newStartTxn)); initialize(newStartTxn); } while ((totalSize + inputData.length) > capacity && !dataMap.isEmpty()) { Map.Entry<Long, byte[]> lowest = dataMap.firstEntry(); dataMap.remove(lowest.getKey()); totalSize -= lowest.getValue().length; } if (inputData.length > capacity) { initialize(INVALID_TXN_ID); Journal.LOG.warn(String.format("A single batch of edits was too " + "large to fit into the cache: startTxn = %d, endTxn = %d, " + "input length = %d. The capacity of the cache (%s) must be " + "increased for it to work properly (current capacity %d)." + "Cache is now empty.", newStartTxn, newEndTxn, inputData.length, DFSConfigKeys.DFS_JOURNALNODE_EDIT_CACHE_SIZE_KEY, capacity)); return; } if (dataMap.isEmpty()) { lowestTxnId = newStartTxn; } else { lowestTxnId = dataMap.firstKey(); } dataMap.put(newStartTxn, inputData); highestTxnId = newEndTxn; totalSize += inputData.length; } } JournaledEditsCache(Configuration conf);   }
@Test public void testCacheSingleSegment() throws Exception { storeEdits(1, 20); assertTxnCountAndContents(1, 5, 5); assertTxnCountAndContents(1, 20, 20); assertTxnCountAndContents(1, 40, 20); assertTxnCountAndContents(10, 11, 20); assertTxnCountAndContents(10, 20, 20); }
@Test public void testCacheBelowCapacityRequestOnBoundary() throws Exception { storeEdits(1, 5); storeEdits(6, 20); storeEdits(21, 30); assertTxnCountAndContents(1, 3, 3); assertTxnCountAndContents(6, 10, 15); assertTxnCountAndContents(1, 7, 7); assertTxnCountAndContents(1, 25, 25); assertTxnCountAndContents(6, 20, 25); assertTxnCountAndContents(6, 50, 30); assertTxnCountAndContents(21, 20, 30); }