AbstractThroughputFileInputOperator extends AbstractFileInputOperator<T> { @Override @SuppressWarnings("unchecked") public Response processStats(BatchedOperatorStats batchedOperatorStats) { BasicCounters<MutableLong> fileCounters = null; for (OperatorStats operatorStats : batchedOperatorStats.getLastWindowedStats()) { if (operatorStats.counters != null) { fileCounters = (BasicCounters<MutableLong>)operatorStats.counters; } } Response response = new Response(); if (fileCounters != null && fileCounters.getCounter(FileCounters.PENDING_FILES).longValue() > 0L || System.currentTimeMillis() - repartitionInterval <= lastRepartition) { response.repartitionRequired = false; return response; } response.repartitionRequired = true; return response; }  void setRepartitionInterval(long repartitionInterval); long getRepartitionInterval(); void setPreferredMaxPendingFilesPerOperator(int pendingFilesPerOperator); int getPreferredMaxPendingFilesPerOperator(); @Override int getPartitionCount(); @Override void setPartitionCount(int requiredPartitions); @Override void emitTuples(); @Override @SuppressWarnings("unchecked") Response processStats(BatchedOperatorStats batchedOperatorStats);  }
@Test public void testProcessStats() throws Exception { underTest.setPartitionCount(4); underTest.setPreferredMaxPendingFilesPerOperator(10); for (int i = 0; i < 21; i++) { underTest.pendingFiles.add("file-" + i); } mockOperatorStats.counters = fileCountersMock; when(mockPartition.getStats()).thenReturn(mockBatchStats); when(mockBatchStats.getLastWindowedStats()).thenReturn(Collections.singletonList(mockOperatorStats)); when(fileCountersMock.getCounter(any(FileCounters.class))).thenReturn(fileCounterMock); when(fileCounterMock.getValue()).thenReturn(20L); Response response = underTest.processStats(mockBatchStats); Assert.assertTrue(response.repartitionRequired); }