### Question:
StageExecution implements StageControl, AutoCloseable { public Iterable<Pair<Step<?>,Float>> stepsOrderedBy( final Key stat, final boolean trueForAscending ) { final List<Step<?>> steps = new ArrayList<>( pipeline ); steps.sort( ( o1, o2 ) -> { Long stat1 = o1.stats().stat( stat ).asLong(); Long stat2 = o2.stats().stat( stat ).asLong(); return trueForAscending ? stat1.compareTo( stat2 ) : stat2.compareTo( stat1 ); } ); return () -> new PrefetchingIterator<Pair<Step<?>,Float>>() { private final Iterator<Step<?>> source = steps.iterator(); private Step<?> next = source.hasNext() ? source.next() : null; @Override protected Pair<Step<?>,Float> fetchNextOrNull() { if ( next == null ) { return null; } Step<?> current = next; next = source.hasNext() ? source.next() : null; float factor = next != null ? (float) stat( current, stat ) / (float) stat( next, stat ) : 1.0f; return Pair.of( current, factor ); } private long stat( Step<?> step, Key stat12 ) { return step.stats().stat( stat12 ).asLong(); } }; } StageExecution( String stageName, String part, Configuration config, Collection<Step<?>> pipeline,
            int orderingGuarantees ); boolean stillExecuting(); void start(); String getStageName(); String name(); Configuration getConfig(); Iterable<Step<?>> steps(); Iterable<Pair<Step<?>,Float>> stepsOrderedBy( final Key stat, final boolean trueForAscending ); int size(); @Override synchronized void panic( Throwable cause ); @Override void assertHealthy(); @Override String toString(); @Override void recycle( Object batch ); @Override T reuse( Supplier<T> fallback ); @Override void close();  }### Answer:
@Test public void shouldOrderStepsAscending() { Collection<Step<?>> steps = new ArrayList<>(); steps.add( stepWithAverageOf( "step1", 0, 10 ) ); steps.add( stepWithAverageOf( "step2", 0, 5 ) ); steps.add( stepWithAverageOf( "step3", 0, 30 ) ); StageExecution execution = new StageExecution( "Test", null, DEFAULT, steps, ORDER_SEND_DOWNSTREAM ); Iterator<Pair<Step<?>,Float>> ordered = execution.stepsOrderedBy( Keys.avg_processing_time, true ).iterator(); Pair<Step<?>,Float> fastest = ordered.next(); assertEquals( 1f / 2f, fastest.other().floatValue(), 0f ); Pair<Step<?>,Float> faster = ordered.next(); assertEquals( 1f / 3f, faster.other().floatValue(), 0f ); Pair<Step<?>,Float> fast = ordered.next(); assertEquals( 1f, fast.other().floatValue(), 0f ); assertFalse( ordered.hasNext() ); }
@Test public void shouldOrderStepsDescending() { Collection<Step<?>> steps = new ArrayList<>(); steps.add( stepWithAverageOf( "step1", 0, 10 ) ); steps.add( stepWithAverageOf( "step2", 0, 5 ) ); steps.add( stepWithAverageOf( "step3", 0, 30 ) ); steps.add( stepWithAverageOf( "step4", 0, 5 ) ); StageExecution execution = new StageExecution( "Test", null, DEFAULT, steps, ORDER_SEND_DOWNSTREAM ); Iterator<Pair<Step<?>,Float>> ordered = execution.stepsOrderedBy( Keys.avg_processing_time, false ).iterator(); Pair<Step<?>,Float> slowest = ordered.next(); assertEquals( 3f, slowest.other().floatValue(), 0f ); Pair<Step<?>,Float> slower = ordered.next(); assertEquals( 2f, slower.other().floatValue(), 0f ); Pair<Step<?>,Float> slow = ordered.next(); assertEquals( 1f, slow.other().floatValue(), 0f ); Pair<Step<?>,Float> alsoSlow = ordered.next(); assertEquals( 1f, alsoSlow.other().floatValue(), 0f ); assertFalse( ordered.hasNext() ); }