@Test public void testGetDirectoriesBefore() throws InterruptedException { final RepositoryConfiguration config = createConfig(2); config.setDesiredIndexSize(4096 * 128); final File storageDir = config.getStorageDirectories().get("1"); final File index1 = new File(storageDir, "index-1"); final File index2 = new File(storageDir, "index-2"); final File[] allIndices = new File[] {index1, index2}; for (final File file : allIndices) { if (file.exists()) { assertTrue(file.delete()); } } assertTrue(index1.mkdirs()); Thread.sleep(1500L); final long timestamp = System.currentTimeMillis(); assertTrue(index2.mkdirs()); try { final IndexDirectoryManager mgr = new IndexDirectoryManager(config); mgr.initialize(); final List<File> dirsBefore = mgr.getDirectoriesBefore(timestamp); assertEquals(1, dirsBefore.size()); assertEquals(index1, dirsBefore.get(0)); } finally { for (final File file : allIndices) { file.delete(); } } }
public synchronized List<File> getDirectoriesBefore(final long timestamp) { final List<File> selected = new ArrayList<>(); final Map<String, List<IndexLocation>> startTimeWithFileByStorageDirectory = flattenDirectoriesByTimestamp().stream() .collect(Collectors.groupingBy(indexLoc -> indexLoc.getPartitionName())); for (final List<IndexLocation> locationList : startTimeWithFileByStorageDirectory.values()) { for (int i = 0; i < locationList.size(); i++) { final IndexLocation indexLoc = locationList.get(i); final String partition = indexLoc.getPartitionName(); final IndexLocation activeLocation = activeIndices.get(partition); if (indexLoc.equals(activeLocation)) { continue; } final Long indexStartTime = indexLoc.getIndexStartTimestamp(); if (indexStartTime > timestamp) { break; } final long indexEndTime = indexLoc.getIndexEndTimestamp(); if (indexEndTime <= timestamp) { logger.debug("Considering Index Location {} older than {} ({}) because its events have an EventTime " + "ranging from {} ({}) to {} ({}) based on the following IndexLocations: {}", indexLoc, timestamp, new Date(timestamp), indexStartTime, new Date(indexStartTime), indexEndTime, new Date(indexEndTime), locationList); selected.add(indexLoc.getIndexDirectory()); } } } logger.debug("Returning the following list of index locations because they were finished being written to before {}: {}", timestamp, selected); return selected; }
IndexDirectoryManager { public synchronized List<File> getDirectoriesBefore(final long timestamp) { final List<File> selected = new ArrayList<>(); final Map<String, List<IndexLocation>> startTimeWithFileByStorageDirectory = flattenDirectoriesByTimestamp().stream() .collect(Collectors.groupingBy(indexLoc -> indexLoc.getPartitionName())); for (final List<IndexLocation> locationList : startTimeWithFileByStorageDirectory.values()) { for (int i = 0; i < locationList.size(); i++) { final IndexLocation indexLoc = locationList.get(i); final String partition = indexLoc.getPartitionName(); final IndexLocation activeLocation = activeIndices.get(partition); if (indexLoc.equals(activeLocation)) { continue; } final Long indexStartTime = indexLoc.getIndexStartTimestamp(); if (indexStartTime > timestamp) { break; } final long indexEndTime = indexLoc.getIndexEndTimestamp(); if (indexEndTime <= timestamp) { logger.debug("Considering Index Location {} older than {} ({}) because its events have an EventTime " + "ranging from {} ({}) to {} ({}) based on the following IndexLocations: {}", indexLoc, timestamp, new Date(timestamp), indexStartTime, new Date(indexStartTime), indexEndTime, new Date(indexEndTime), locationList); selected.add(indexLoc.getIndexDirectory()); } } } logger.debug("Returning the following list of index locations because they were finished being written to before {}: {}", timestamp, selected); return selected; } }
IndexDirectoryManager { public synchronized List<File> getDirectoriesBefore(final long timestamp) { final List<File> selected = new ArrayList<>(); final Map<String, List<IndexLocation>> startTimeWithFileByStorageDirectory = flattenDirectoriesByTimestamp().stream() .collect(Collectors.groupingBy(indexLoc -> indexLoc.getPartitionName())); for (final List<IndexLocation> locationList : startTimeWithFileByStorageDirectory.values()) { for (int i = 0; i < locationList.size(); i++) { final IndexLocation indexLoc = locationList.get(i); final String partition = indexLoc.getPartitionName(); final IndexLocation activeLocation = activeIndices.get(partition); if (indexLoc.equals(activeLocation)) { continue; } final Long indexStartTime = indexLoc.getIndexStartTimestamp(); if (indexStartTime > timestamp) { break; } final long indexEndTime = indexLoc.getIndexEndTimestamp(); if (indexEndTime <= timestamp) { logger.debug("Considering Index Location {} older than {} ({}) because its events have an EventTime " + "ranging from {} ({}) to {} ({}) based on the following IndexLocations: {}", indexLoc, timestamp, new Date(timestamp), indexStartTime, new Date(indexStartTime), indexEndTime, new Date(indexEndTime), locationList); selected.add(indexLoc.getIndexDirectory()); } } } logger.debug("Returning the following list of index locations because they were finished being written to before {}: {}", timestamp, selected); return selected; } IndexDirectoryManager(final RepositoryConfiguration repoConfig); }
IndexDirectoryManager { public synchronized List<File> getDirectoriesBefore(final long timestamp) { final List<File> selected = new ArrayList<>(); final Map<String, List<IndexLocation>> startTimeWithFileByStorageDirectory = flattenDirectoriesByTimestamp().stream() .collect(Collectors.groupingBy(indexLoc -> indexLoc.getPartitionName())); for (final List<IndexLocation> locationList : startTimeWithFileByStorageDirectory.values()) { for (int i = 0; i < locationList.size(); i++) { final IndexLocation indexLoc = locationList.get(i); final String partition = indexLoc.getPartitionName(); final IndexLocation activeLocation = activeIndices.get(partition); if (indexLoc.equals(activeLocation)) { continue; } final Long indexStartTime = indexLoc.getIndexStartTimestamp(); if (indexStartTime > timestamp) { break; } final long indexEndTime = indexLoc.getIndexEndTimestamp(); if (indexEndTime <= timestamp) { logger.debug("Considering Index Location {} older than {} ({}) because its events have an EventTime " + "ranging from {} ({}) to {} ({}) based on the following IndexLocations: {}", indexLoc, timestamp, new Date(timestamp), indexStartTime, new Date(indexStartTime), indexEndTime, new Date(indexEndTime), locationList); selected.add(indexLoc.getIndexDirectory()); } } } logger.debug("Returning the following list of index locations because they were finished being written to before {}: {}", timestamp, selected); return selected; } IndexDirectoryManager(final RepositoryConfiguration repoConfig); synchronized void initialize(); synchronized void deleteDirectory(final File directory); synchronized List<File> getDirectoriesBefore(final long timestamp); synchronized List<File> getDirectories(final Long startTime, final Long endTime); synchronized List<File> getDirectories(final Long startTime, final Long endTime, final String partitionName); boolean onIndexCommitted(final File indexDir); synchronized Optional<File> getActiveIndexDirectory(final String partitionName); synchronized File getWritableIndexingDirectory(final long earliestTimestamp, final String partitionName); }
IndexDirectoryManager { public synchronized List<File> getDirectoriesBefore(final long timestamp) { final List<File> selected = new ArrayList<>(); final Map<String, List<IndexLocation>> startTimeWithFileByStorageDirectory = flattenDirectoriesByTimestamp().stream() .collect(Collectors.groupingBy(indexLoc -> indexLoc.getPartitionName())); for (final List<IndexLocation> locationList : startTimeWithFileByStorageDirectory.values()) { for (int i = 0; i < locationList.size(); i++) { final IndexLocation indexLoc = locationList.get(i); final String partition = indexLoc.getPartitionName(); final IndexLocation activeLocation = activeIndices.get(partition); if (indexLoc.equals(activeLocation)) { continue; } final Long indexStartTime = indexLoc.getIndexStartTimestamp(); if (indexStartTime > timestamp) { break; } final long indexEndTime = indexLoc.getIndexEndTimestamp(); if (indexEndTime <= timestamp) { logger.debug("Considering Index Location {} older than {} ({}) because its events have an EventTime " + "ranging from {} ({}) to {} ({}) based on the following IndexLocations: {}", indexLoc, timestamp, new Date(timestamp), indexStartTime, new Date(indexStartTime), indexEndTime, new Date(indexEndTime), locationList); selected.add(indexLoc.getIndexDirectory()); } } } logger.debug("Returning the following list of index locations because they were finished being written to before {}: {}", timestamp, selected); return selected; } IndexDirectoryManager(final RepositoryConfiguration repoConfig); synchronized void initialize(); synchronized void deleteDirectory(final File directory); synchronized List<File> getDirectoriesBefore(final long timestamp); synchronized List<File> getDirectories(final Long startTime, final Long endTime); synchronized List<File> getDirectories(final Long startTime, final Long endTime, final String partitionName); boolean onIndexCommitted(final File indexDir); synchronized Optional<File> getActiveIndexDirectory(final String partitionName); synchronized File getWritableIndexingDirectory(final long earliestTimestamp, final String partitionName);  }