AbstractBlockReader extends BaseOperator implements Partitioner<AbstractBlockReader<R, B, STREAM>>, StatsListener, Operator.IdleTimeHandler { @Override public Response processStats(BatchedOperatorStats stats) { response.repartitionRequired = false; if (!collectStats) { return response; } List<Stats.OperatorStats> lastWindowedStats = stats.getLastWindowedStats(); if (lastWindowedStats != null && lastWindowedStats.size() > 0) { Stats.OperatorStats lastStats = lastWindowedStats.get(lastWindowedStats.size() - 1); if (lastStats.inputPorts.size() > 0) { backlogPerOperator.put(stats.getOperatorId(), lastStats.inputPorts.get(0).queueSize); } } if (System.currentTimeMillis() < nextMillis) { return response; } nextMillis = System.currentTimeMillis() + intervalMillis; LOG.debug("Proposed NextMillis = {}", nextMillis); long totalBacklog = 0; for (Map.Entry<Integer, Integer> backlog : backlogPerOperator.entrySet()) { totalBacklog += backlog.getValue(); } LOG.debug("backlog {} partitionCount {}", totalBacklog, partitionCount); backlogPerOperator.clear(); if (totalBacklog == partitionCount) { return response; } int newPartitionCount; if (totalBacklog > maxReaders) { LOG.debug("large backlog {}", totalBacklog); newPartitionCount = maxReaders; } else if (totalBacklog < minReaders) { LOG.debug("small backlog {}", totalBacklog); newPartitionCount = minReaders; } else { newPartitionCount = getAdjustedCount(totalBacklog); LOG.debug("moderate backlog {}", totalBacklog); } LOG.debug("backlog {} newPartitionCount {} partitionCount {}", totalBacklog, newPartitionCount, partitionCount); if (newPartitionCount == partitionCount) { return response; } partitionCount = newPartitionCount; response.repartitionRequired = true; LOG.debug("partition required", totalBacklog, partitionCount); return response; } 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 testProcessStatsForPartitionCount() { TestUtils.MockBatchedOperatorStats readerStats = new TestUtils.MockBatchedOperatorStats(2); readerStats.operatorStats = Lists.newArrayList(); readerStats.operatorStats.add(new ReaderStats(10, 1, 100, 1)); TestReader sliceReader = new TestReader(); StatsListener.Response response = sliceReader.processStats(readerStats); Assert.assertTrue("partition needed", response.repartitionRequired); Assert.assertEquals("partition count changed", 8, sliceReader.getPartitionCount()); }