StripWeakCentroidsRunner implements MapReduceJobRunner { @Override public int run(final Configuration config, final PropertyManagement runTimeProperties) throws Exception { currentCentroidCount = 0; final CentroidManager<T> centroidManager = constructCentroidManager(config, runTimeProperties); return centroidManager.processForAllGroups(new CentroidProcessingFn<T>() { @Override public int processGroup(final String groupID, final List<AnalyticItemWrapper<T>> centroids) { if (centroids.size() <= minimum) { currentCentroidCount = centroids.size(); return 0; } Collections.sort(centroids, new Comparator<AnalyticItemWrapper<T>>() { @Override public int compare(final AnalyticItemWrapper<T> arg0, final AnalyticItemWrapper<T> arg1) { return (arg1.getAssociationCount() - arg0.getAssociationCount()) < 0 ? -1 : 1; } }); int position = breakStrategy.getBreakPoint(centroids); position = Math.min(Math.max(minimum, position), maximum); final String toDelete[] = new String[centroids.size() - position]; LOGGER.info("Deleting {} out of {}", toDelete.length, centroids.size()); int count = 0; final Iterator<AnalyticItemWrapper<T>> it = centroids.iterator(); while (it.hasNext()) { final AnalyticItemWrapper<T> centroid = it.next(); if (count++ >= position) { toDelete[count - position - 1] = centroid.getID(); } } try { centroidManager.delete(toDelete); } catch (final IOException e) { LOGGER.warn("Unable to delete the centriod mamager", e); return -1; } currentCentroidCount += position; return 0; } }); } StripWeakCentroidsRunner(); void setBreakStrategy(final BreakStrategy<T> breakStrategy); void setRange(final int minimum, final int maximum); int getCurrentCentroidCount(); @Override int run(final Configuration config, final PropertyManagement runTimeProperties);  }
@Test public void testCliff() throws Exception { final StripWeakCentroidsRunnerForTestOne testObj = new StripWeakCentroidsRunnerForTestOne(); testObj.run(new Configuration(), new PropertyManagement()); }