test_class.declarator: EMPTY_DATA = new byte[0]
test_class.modifier: private static final
test_class.original_string: private static final byte[] EMPTY_DATA = new byte[0];
test_class.type: byte[]
test_class.var_name: EMPTY_DATA
test_class.declarator: queue
test_class.modifier: private
test_class.original_string: private SegmentWriteQueue queue;
test_class.type: SegmentWriteQueue
test_class.var_name: queue
test_class.declarator: queueBlocked
test_class.modifier: private
test_class.original_string: private SegmentWriteQueue queueBlocked;
test_class.type: SegmentWriteQueue
test_class.var_name: queueBlocked
test_class.file: oak-segment-remote/src/test/java/org/apache/jackrabbit/oak/segment/remote/queue/SegmentWriteQueueTest.java
test_class.identifier: SegmentWriteQueueTest
test_case.body: @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)));
          }
      }
test_case.class_method_signature: SegmentWriteQueueTest.testFlush()
test_case.full_signature: @Test(timeout = 1000) public void testFlush()
test_case.identifier: testFlush
test_case.invocations: synchronizedSet acquire add getMsb getLsb addToQueue tarEntry start flush set addAll sleep assertFalse get start addToQueue tarEntry set sleep assertFalse get release get sleep assertTrue get assertTrue get assertTrue contains uuid
test_case.modifiers: @Test(timeout = 1000) public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testFlush()
focal_class.declarator: THREADS = Integer.getInteger("oak.segment.remote.threads", 5)
focal_class.modifier: public static final
focal_class.original_string: public static final int THREADS = Integer.getInteger("oak.segment.remote.threads", 5);
focal_class.type: int
focal_class.var_name: THREADS
focal_class.declarator: QUEUE_SIZE = Integer.getInteger("oak.segment.remote.queue.size", 20)
focal_class.modifier: private static final
focal_class.original_string: private static final int QUEUE_SIZE = Integer.getInteger("oak.segment.remote.queue.size", 20);
focal_class.type: int
focal_class.var_name: QUEUE_SIZE
focal_class.declarator: log = LoggerFactory.getLogger(SegmentWriteQueue.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(SegmentWriteQueue.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: queue
focal_class.modifier: private final
focal_class.original_string: private final BlockingDeque<SegmentWriteAction> queue;
focal_class.type: BlockingDeque<SegmentWriteAction>
focal_class.var_name: queue
focal_class.declarator: segmentsByUUID
focal_class.modifier: private final
focal_class.original_string: private final Map<UUID, SegmentWriteAction> segmentsByUUID;
focal_class.type: Map<UUID, SegmentWriteAction>
focal_class.var_name: segmentsByUUID
focal_class.declarator: executor
focal_class.modifier: private final
focal_class.original_string: private final ExecutorService executor;
focal_class.type: ExecutorService
focal_class.var_name: executor
focal_class.declarator: flushLock
focal_class.modifier: private final
focal_class.original_string: private final ReadWriteLock flushLock;
focal_class.type: ReadWriteLock
focal_class.var_name: flushLock
focal_class.declarator: writer
focal_class.modifier: private final
focal_class.original_string: private final SegmentConsumer writer;
focal_class.type: SegmentConsumer
focal_class.var_name: writer
focal_class.declarator: shutdown
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean shutdown;
focal_class.type: boolean
focal_class.var_name: shutdown
focal_class.declarator: brokenMonitor = new Object()
focal_class.modifier: private final
focal_class.original_string: private final Object brokenMonitor = new Object();
focal_class.type: Object
focal_class.var_name: brokenMonitor
focal_class.declarator: broken
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean broken;
focal_class.type: boolean
focal_class.var_name: broken
focal_class.file: oak-segment-remote/src/main/java/org/apache/jackrabbit/oak/segment/remote/queue/SegmentWriteQueue.java
focal_class.identifier: SegmentWriteQueue
focal_class.interfaces: implements Closeable
focal_class.class_method_signature: SegmentWriteQueue.SegmentWriteQueue(SegmentConsumer writer)
focal_class.full_signature: public  SegmentWriteQueue(SegmentConsumer writer)
focal_class.identifier: SegmentWriteQueue
focal_class.modifiers: public
focal_class.parameters: (SegmentConsumer writer)
focal_class.signature:  SegmentWriteQueue(SegmentConsumer writer)
focal_class.class_method_signature: SegmentWriteQueue.SegmentWriteQueue(SegmentConsumer writer, int queueSize, int threadNo)
focal_class.full_signature:   SegmentWriteQueue(SegmentConsumer writer, int queueSize, int threadNo)
focal_class.identifier: SegmentWriteQueue
focal_class.parameters: (SegmentConsumer writer, int queueSize, int threadNo)
focal_class.signature:  SegmentWriteQueue(SegmentConsumer writer, int queueSize, int threadNo)
focal_class.class_method_signature: SegmentWriteQueue.mainLoop()
focal_class.full_signature: private void mainLoop()
focal_class.identifier: mainLoop
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void mainLoop()
focal_class.class_method_signature: SegmentWriteQueue.consume()
focal_class.full_signature: private void consume()
focal_class.identifier: consume
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void consume()
focal_class.class_method_signature: SegmentWriteQueue.consume(SegmentWriteAction segment)
focal_class.full_signature: private void consume(SegmentWriteAction segment)
focal_class.identifier: consume
focal_class.modifiers: private
focal_class.parameters: (SegmentWriteAction segment)
focal_class.return: void
focal_class.signature: void consume(SegmentWriteAction segment)
focal_class.class_method_signature: SegmentWriteQueue.emergencyLoop()
focal_class.full_signature: private void emergencyLoop()
focal_class.identifier: emergencyLoop
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void emergencyLoop()
focal_class.class_method_signature: SegmentWriteQueue.addToQueue(RemoteSegmentArchiveEntry indexEntry, byte[] data, int offset, int size)
focal_class.full_signature: public void addToQueue(RemoteSegmentArchiveEntry indexEntry, byte[] data, int offset, int size)
focal_class.identifier: addToQueue
focal_class.modifiers: public
focal_class.parameters: (RemoteSegmentArchiveEntry indexEntry, byte[] data, int offset, int size)
focal_class.return: void
focal_class.signature: void addToQueue(RemoteSegmentArchiveEntry indexEntry, byte[] data, int offset, int size)
focal_class.class_method_signature: SegmentWriteQueue.flush()
focal_class.full_signature: public void flush()
focal_class.identifier: flush
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void flush()
focal_class.class_method_signature: SegmentWriteQueue.read(UUID id)
focal_class.full_signature: public SegmentWriteAction read(UUID id)
focal_class.identifier: read
focal_class.modifiers: public
focal_class.parameters: (UUID id)
focal_class.return: SegmentWriteAction
focal_class.signature: SegmentWriteAction read(UUID id)
focal_class.class_method_signature: SegmentWriteQueue.close()
focal_class.full_signature: @Override public void close()
focal_class.identifier: close
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.class_method_signature: SegmentWriteQueue.isEmpty()
focal_class.full_signature: public boolean isEmpty()
focal_class.identifier: isEmpty
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean isEmpty()
focal_class.class_method_signature: SegmentWriteQueue.isBroken()
focal_class.full_signature:  boolean isBroken()
focal_class.identifier: isBroken
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean isBroken()
focal_class.class_method_signature: SegmentWriteQueue.getSize()
focal_class.full_signature:  int getSize()
focal_class.identifier: getSize
focal_class.parameters: ()
focal_class.return: int
focal_class.signature: int getSize()
focal_class.class_method_signature: SegmentWriteQueue.setBroken(boolean broken)
focal_class.full_signature: private void setBroken(boolean broken)
focal_class.identifier: setBroken
focal_class.modifiers: private
focal_class.parameters: (boolean broken)
focal_class.return: void
focal_class.signature: void setBroken(boolean broken)
focal_class.class_method_signature: SegmentWriteQueue.waitWhileBroken()
focal_class.full_signature: private void waitWhileBroken()
focal_class.identifier: waitWhileBroken
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void waitWhileBroken()
focal_class.class_method_signature: SegmentWriteQueue.waitUntilBroken()
focal_class.full_signature: private void waitUntilBroken()
focal_class.identifier: waitUntilBroken
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void waitUntilBroken()
focal_method.body: 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();
          }
      }
focal_method.class_method_signature: SegmentWriteQueue.flush()
focal_method.full_signature: public void flush()
focal_method.identifier: flush
focal_method.invocations: lock writeLock currentTimeMillis isEmpty wait currentTimeMillis toMillis error currentTimeMillis unlock writeLock
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void flush()