SegmentWriteQueue implements Closeable { public void flush() throws IOException { flushLock.writeLock().lock(); try { synchronized (segmentsByUUID) { long start = System.currentTimeMillis(); while (!segmentsByUUID.isEmpty()) { segmentsByUUID.wait(100); if (System.currentTimeMillis() - start > TimeUnit.MINUTES.toMillis(1)) { log.error("Can't flush the queue in 1 minute. Queue: {}. Segment map: {}", queue, segmentsByUUID); start = System.currentTimeMillis(); } } } } catch (InterruptedException e) { throw new IOException(e); } finally { flushLock.writeLock().unlock(); } } SegmentWriteQueue(SegmentConsumer writer);   SegmentWriteQueue(SegmentConsumer writer, int queueSize, int threadNo); void addToQueue(RemoteSegmentArchiveEntry indexEntry, byte[] data, int offset, int size); void flush(); SegmentWriteAction read(UUID id); @Override void close(); boolean isEmpty(); static final int THREADS; }
@Test(timeout = 1000) public void testFlush() throws IOException, InterruptedException { Set<UUID> added = Collections.synchronizedSet(new HashSet<>()); Semaphore semaphore = new Semaphore(0); queue = new SegmentWriteQueue((tarEntry, data, offset, size) -> { try { semaphore.acquire(); } catch (InterruptedException e) { } added.add(new UUID(tarEntry.getMsb(), tarEntry.getLsb())); }); for (int i = 0; i < 3; i++) { queue.addToQueue(tarEntry(i), EMPTY_DATA, 0, 0); } AtomicBoolean flushFinished = new AtomicBoolean(false); Set<UUID> addedAfterFlush = new HashSet<>(); new Thread(() -> { try { queue.flush(); flushFinished.set(true); addedAfterFlush.addAll(added); } catch (IOException e) { throw new UncheckedIOException(e); } }).start(); Thread.sleep(100); assertFalse("Flush should be blocked", flushFinished.get()); AtomicBoolean addFinished = new AtomicBoolean(false); new Thread(() -> { try { queue.addToQueue(tarEntry(10), EMPTY_DATA, 0, 0); addFinished.set(true); } catch (IOException e) { throw new UncheckedIOException(e); } }).start(); Thread.sleep(100); assertFalse("Adding segments should be blocked until the flush is finished", addFinished.get()); semaphore.release(Integer.MAX_VALUE); while (!addFinished.get()) { Thread.sleep(10); } assertTrue("Flush should be finished once the ", flushFinished.get()); assertTrue("Adding segments should be blocked until the flush is finished", addFinished.get()); for (int i = 0; i < 3; i++) { assertTrue(addedAfterFlush.contains(uuid(i))); } }