AbstractBlockReader extends BaseOperator implements Partitioner<AbstractBlockReader<R, B, STREAM>>, StatsListener, Operator.IdleTimeHandler { protected int getAdjustedCount(long newCount) { int adjustCount = 1; while (adjustCount < newCount) { adjustCount <<= 1; } if (adjustCount > newCount) { adjustCount >>>= 1; } LOG.debug("adjust {} => {}", newCount, adjustCount); return adjustCount; } AbstractBlockReader(); @Override void setup(Context.OperatorContext context); @Override void beginWindow(long windowId); @Override void handleIdleTime(); @Override void endWindow(); @SuppressWarnings("unchecked") @Override Collection<Partition<AbstractBlockReader<R, B, STREAM>>> definePartitions(
      Collection<Partition<AbstractBlockReader<R, B, STREAM>>> partitions, PartitioningContext context); @Override void partitioned(Map<Integer, Partition<AbstractBlockReader<R, B, STREAM>>> integerPartitionMap); @Override Response processStats(BatchedOperatorStats stats); void setMaxReaders(int maxReaders); void setMinReaders(int minReaders); int getMaxReaders(); int getMinReaders(); void setCollectStats(boolean collectStats); boolean isCollectStats(); void setIntervalMillis(long intervalMillis); long getIntervalMillis(); void setReaderContext(ReaderContext<STREAM> readerContext); ReaderContext<STREAM> getReaderContext(); @Override String toString(); final transient DefaultOutputPort<B> blocksMetadataOutput; final transient DefaultOutputPort<ReaderRecord<R>> messages; final transient DefaultInputPort<B> blocksMetadataInput; }
@Test public void testAdjustedCount() { TestReader sliceReader = new TestReader(); Assert.assertEquals("min", 1, sliceReader.getAdjustedCount(1)); Assert.assertEquals("max", 16, sliceReader.getAdjustedCount(16)); Assert.assertEquals("max-1", 8, sliceReader.getAdjustedCount(15)); Assert.assertEquals("min+1", 2, sliceReader.getAdjustedCount(2)); Assert.assertEquals("between 1", 4, sliceReader.getAdjustedCount(4)); Assert.assertEquals("between 2", 4, sliceReader.getAdjustedCount(7)); Assert.assertEquals("between 2", 8, sliceReader.getAdjustedCount(12)); }