region StreamSegmentStore Implementation
8937
@Override
    public CompletableFuture<Void> append(String streamSegmentName, byte[] data, Collection<AttributeUpdate> attributeUpdates, Duration timeout) {
        ensureRunning();

        TimeoutTimer timer = new TimeoutTimer(timeout);
        logRequest("append", streamSegmentName, data.length);
        this.metrics.append();
        return this.metadataStore.getOrAssignSegmentId(streamSegmentName, timer.getRemaining(),
                streamSegmentId -> {
                    StreamSegmentAppendOperation operation = new StreamSegmentAppendOperation(streamSegmentId, data, attributeUpdates);
                    return processAttributeUpdaterOperation(operation, timer);
                });
    }
62962
public void reconstituteStreamSet(StreamSet streamSet)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this, tc, "reconstituteStreamSet", streamSet);

    synchronized(streamSets)
    {
      streamSets.put(streamSet.getStreamID(), streamSet);
      sourceMap.put(streamSet.getStreamID(), streamSet.getRemoteMEUuid());
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "reconstituteStreamSet");
  }
706171
Map<Integer, StreamState> activeStreams() {
        Map<Integer, StreamState> streams = new TreeMap<Integer, StreamState>(streamComparator);
        streams.putAll(activeStreams);
        return streams;
    }
1185623
private void createNewFiltersAndCursors(ItemStream itemStream) 
    throws SIResourceException, MessageStoreException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "createNewFiltersAndCursors", itemStream);
    
    LockingCursor cursor = null;
 
    // Instantiate a new array of Filters and associated cursors. If there is
    // no message classification, then we'll instantiate a single filter and 
    // cursor pair.
    if(classifyingMessages)
    {
      // Classifying messages for XD.
      // this work should be done under the classifications readlock acquired higher
      // up in the stack
      JSConsumerClassifications classifications = consumerSet.getClassifications();
      int numClasses = classifications.getNumberOfClasses();
      consumerKeyFilter = new LocalQPConsumerKeyFilter[numClasses+1];
      for(int i=0;i<numClasses+1;i++)
      {
        String classificationName = null;
        // The zeroth filter belongs to the default classification, which has a 
        // null classification name
        if(i > 0)
          classificationName = classifications.getClassification(i);
        consumerKeyFilter[i] = new LocalQPConsumerKeyFilter(this, i, classificationName);
        cursor = itemStream.newLockingItemCursor(consumerKeyFilter[i], !forwardScanning);
        consumerKeyFilter[i].setLockingCursor(cursor);
      }
    }
    else
    {
      // No message classification
      consumerKeyFilter = new LocalQPConsumerKeyFilter[1];
      consumerKeyFilter[0] = new LocalQPConsumerKeyFilter(this, 0, null);
      cursor = itemStream.newLockingItemCursor(consumerKeyFilter[0], !forwardScanning);
      consumerKeyFilter[0].setLockingCursor(cursor);
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "createNewFiltersAndCursors");
  }
653243
public void createShellStreamSet()
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "createShellStreamSet");

        StreamSet streamSet = null;

        synchronized (streamSets)
        {
            streamSet = new StreamSet(null,
                            targetMEUuid,
                            0,
                            isLink ? StreamSet.Type.LINK_INTERNAL_OUTPUT : StreamSet.Type.INTERNAL_OUTPUT);
            streamSets.put(null, streamSet);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "createShellStreamSet", streamSet);
    }
1211813
private <T> Collection<Integer> transformPartition(PartitionTransformation<T> partition) {
		StreamTransformation<T> input = partition.getInput();
		List<Integer> resultIds = new ArrayList<>();

		Collection<Integer> transformedIds = transform(input);
		for (Integer transformedId: transformedIds) {
			int virtualId = StreamTransformation.getNewNodeId();
			streamGraph.addVirtualPartitionNode(transformedId, virtualId, partition.getPartitioner());
			resultIds.add(virtualId);
		}

		return resultIds;
	}
1266612
public void setStreams(java.util.Collection<Stream> streams) {
        if (streams == null) {
            this.streams = null;
            return;
        }

        this.streams = new java.util.ArrayList<Stream>(streams);
    }
1107354
private StreamSegmentReadIndex getOrCreateIndex(long streamSegmentId) throws StreamSegmentNotExistsException {
        StreamSegmentReadIndex index;
        synchronized (this.lock) {
            // Try to see if we have the index already in memory.
            index = getIndex(streamSegmentId);
            if (index == null) {
                // We don't have it, create one.
                SegmentMetadata segmentMetadata = this.metadata.getStreamSegmentMetadata(streamSegmentId);
                Exceptions.checkArgument(segmentMetadata != null, "streamSegmentId",
                        "StreamSegmentId %s does not exist in the metadata.", streamSegmentId);
                if (segmentMetadata.isDeleted()) {
                    throw new StreamSegmentNotExistsException(segmentMetadata.getName());
                }

                index = new StreamSegmentReadIndex(this.config, segmentMetadata, this.cache, this.storage, this.executor, isRecoveryMode());
                this.cacheManager.register(index);
                this.readIndices.put(streamSegmentId, index);
            }
        }

        return index;
    }
397336
public static <K, V> Stream<Entry<K, V>> sortedStream(Map<K, V> map,
            Comparator<? super Entry<K, V>> comparator) {
        return buildEntryStream(map).sorted(comparator);
    }
82408
@Override
    public List<Rectangle> getSegmentedRegions(BufferedImage bi, int pageIteratorLevel) throws TesseractException {
        init();
        setTessVariables();

        try {
            List<Rectangle> list = new ArrayList<Rectangle>();
            setImage(bi, null);

            Boxa boxes = TessBaseAPIGetComponentImages(handle, pageIteratorLevel, TRUE, null, null);
            int boxCount = Leptonica1.boxaGetCount(boxes);
            for (int i = 0; i < boxCount; i++) {
                Box box = Leptonica1.boxaGetBox(boxes, i, L_CLONE);
                if (box == null) {
                    continue;
                }
                list.add(new Rectangle(box.x, box.y, box.w, box.h));
                PointerByReference pRef = new PointerByReference();
                pRef.setValue(box.getPointer());
                Leptonica1.boxDestroy(pRef);
            }

            PointerByReference pRef = new PointerByReference();
            pRef.setValue(boxes.getPointer());
            Leptonica1.boxaDestroy(pRef);

            return list;
        } catch (IOException ioe) {
            // skip the problematic image
            logger.warn(ioe.getMessage(), ioe);
            throw new TesseractException(ioe);
        } finally {
            dispose();
        }
    }
207677
public static List<TrackSegment> nonEmptySegments(
		final List<TrackSegment> segments
	) {
		return segments.stream()
			.filter(TrackSegment::nonEmpty)
			.collect(toList());
	}
57450
private synchronized void resetCurrentSegment() {
    Segment lastSegment = lastSegment();
    if (lastSegment != null) {
      currentSegment = lastSegment;
    } else {
      SegmentDescriptor descriptor = SegmentDescriptor.builder()
        .withId(1)
        .withVersion(1)
        .withIndex(1)
        .withMaxSegmentSize(storage.maxSegmentSize())
        .withMaxEntries(storage.maxEntriesPerSegment())
        .build();
      descriptor.lock();

      currentSegment = createSegment(descriptor);

      segments.put(1L, currentSegment);
    }
  }
33530
public <T> StreamSupplier<T> queryRawStream(List<String> dimensions, List<String> storedMeasures, AggregationPredicate where,
			Class<T> resultClass) throws QueryException {
		return queryRawStream(dimensions, storedMeasures, where, resultClass, classLoader);
	}
267957
@GuardedBy("readers")
    private void releaseSegmentsIfNeeded() throws ReaderNotInReaderGroupException {
        releaseSealedSegments();
        Segment segment = groupState.findSegmentToReleaseIfRequired();
        if (segment != null) {
            log.info("{} releasing segment {}", this, segment);
            EventSegmentReader reader = readers.stream().filter(r -> r.getSegmentId().equals(segment)).findAny().orElse(null);
            if (reader != null) {
                if (groupState.releaseSegment(segment, reader.getOffset(), getLag())) {
                    readers.remove(reader);
                    reader.close();
                }
            }
        }
    }
126099
protected void _initSegmentData(int sz) {
		if (m_segmentParamIndex == null) {
			m_segmentFlags = (AttributeStreamOfInt8) AttributeStreamBase
					.createByteStream(m_pointCount,
							(byte) SegmentFlags.enumLineSeg);
			m_segmentParamIndex = (AttributeStreamOfInt32) AttributeStreamBase
					.createIndexStream(m_pointCount, -1);
		}

		int size = m_curveParamwritePoint + sz;
		if (m_segmentParams == null) {
			m_segmentParams = (AttributeStreamOfDbl) AttributeStreamBase
					.createAttributeStreamWithPersistence(
							VertexDescription.Persistence.enumDouble, size);
		} else {
			m_segmentParams.resize(size, 0);
		}
	}
1232390
Map<Segment, Long> acquireNewSegmentsIfNeeded(long timeLag) throws ReaderNotInReaderGroupException {
        fetchUpdatesIfNeeded();
        if (shouldAcquireSegment()) {
            return acquireSegment(timeLag);
        } else {
            return Collections.emptyMap();
        }
    }
253861
private void readSegments()
    throws IOException
  {
    for (SegmentExtent10 extent : _segmentExtents) {
      try (ReadStream is = openRead(extent.address(), extent.length())) {
        is.skip(extent.length() - BLOCK_SIZE);
        
        long sequence = BitsUtil.readLong(is);
        
        byte []tableKey = new byte[TABLE_KEY_SIZE];
        is.readAll(tableKey, 0, tableKey.length);
        
        // XXX: crc
        
        if (sequence > 0) {
          Segment10 segment = new Segment10(sequence, tableKey, extent);
          
          _segments.add(segment);
        }
      }
    }
  }