@@ -255,47 +255,49 @@ public List<DataSegment> getCachedSegments() throws IOException
255255 private void addFilesToCachedSegments (File file , AtomicInteger ignored , List <DataSegment > cachedSegments ) throws IOException
256256 {
257257 final DataSegment segment = jsonMapper .readValue (file , DataSegment .class );
258- boolean removeInfo = false ;
258+
259259 if (!segment .getId ().toString ().equals (file .getName ())) {
260260 log .warn ("Ignoring cache file[%s] for segment[%s]." , file .getPath (), segment .getId ());
261261 ignored .incrementAndGet ();
262- } else {
263- removeInfo = true ;
264- final SegmentCacheEntry cacheEntry = new SegmentCacheEntry (segment );
265- for (StorageLocation location : locations ) {
266- // check for migrate from old nested local storage path format
267- final File legacyPath = new File (location .getPath (), DataSegmentPusher .getDefaultStorageDir (segment , false ));
268- if (legacyPath .exists ()) {
269- final File destination = cacheEntry .toPotentialLocation (location .getPath ());
270- FileUtils .mkdirp (destination );
271- final File [] oldFiles = legacyPath .listFiles ();
272- final File [] newFiles = destination .listFiles ();
273- // make sure old files exist and new files do not exist
274- if (oldFiles != null && oldFiles .length > 0 && newFiles != null && newFiles .length == 0 ) {
275- Files .move (legacyPath .toPath (), destination .toPath (), StandardCopyOption .ATOMIC_MOVE );
276- }
277- cleanupLegacyCacheLocation (location .getPath (), legacyPath );
262+ return ;
263+ }
264+
265+ boolean removeInfo = true ;
266+
267+ final SegmentCacheEntry cacheEntry = new SegmentCacheEntry (segment );
268+ for (StorageLocation location : locations ) {
269+ // check for migrate from old nested local storage path format
270+ final File legacyPath = new File (location .getPath (), DataSegmentPusher .getDefaultStorageDir (segment , false ));
271+ if (legacyPath .exists ()) {
272+ final File destination = cacheEntry .toPotentialLocation (location .getPath ());
273+ FileUtils .mkdirp (destination );
274+ final File [] oldFiles = legacyPath .listFiles ();
275+ final File [] newFiles = destination .listFiles ();
276+ // make sure old files exist and new files do not exist
277+ if (oldFiles != null && oldFiles .length > 0 && newFiles != null && newFiles .length == 0 ) {
278+ Files .move (legacyPath .toPath (), destination .toPath (), StandardCopyOption .ATOMIC_MOVE );
278279 }
280+ cleanupLegacyCacheLocation (location .getPath (), legacyPath );
281+ }
279282
280- if (cacheEntry .checkExists (location .getPath ())) {
281- removeInfo = false ;
282- final boolean reserveResult ;
283- if (config .isVirtualStorage ()) {
284- reserveResult = location .reserveWeak (cacheEntry );
285- } else {
286- reserveResult = location .reserve (cacheEntry );
287- }
288- if (!reserveResult ) {
289- log .makeAlert (
290- "storage[%s:%,d] has more segments than it is allowed. Currently loading Segment[%s:%,d]. Please increase druid.segmentCache.locations maxSize param" ,
291- location .getPath (),
292- location .availableSizeBytes (),
293- segment .getId (),
294- segment .getSize ()
295- ).emit ();
296- }
297- cachedSegments .add (segment );
283+ if (cacheEntry .checkExists (location .getPath ())) {
284+ removeInfo = false ;
285+ final boolean reserveResult ;
286+ if (config .isVirtualStorage ()) {
287+ reserveResult = location .reserveWeak (cacheEntry );
288+ } else {
289+ reserveResult = location .reserve (cacheEntry );
290+ }
291+ if (!reserveResult ) {
292+ log .makeAlert (
293+ "storage[%s:%,d] has more segments than it is allowed. Currently loading Segment[%s:%,d]. Please increase druid.segmentCache.locations maxSize param" ,
294+ location .getPath (),
295+ location .availableSizeBytes (),
296+ segment .getId (),
297+ segment .getSize ()
298+ ).emit ();
298299 }
300+ cachedSegments .add (segment );
299301 }
300302 }
301303
0 commit comments