### Question:
BadCollector implements Collector { @Override public void collectDuplicateNode( final Object id, long actualId, final String group ) { collect( new NodesProblemReporter( id, group ) ); } BadCollector( OutputStream out, long tolerance, int collect );   BadCollector( OutputStream out, long tolerance, int collect, int backPressureThreshold, boolean skipBadEntriesLogging, Monitor monitor ); @Override void collectBadRelationship( Object startId, String startIdGroup, String type, Object endId,
            String endIdGroup, Object specificValue ); @Override void collectExtraColumns( final String source, final long row, final String value ); @Override void collectDuplicateNode( final Object id, long actualId, final String group ); @Override boolean isCollectingBadRelationships(); @Override void close(); @Override long badEntries(); static final long UNLIMITED_TOLERANCE; }### Answer:
@Test public void shouldApplyBackPressure() throws Exception { int backPressureThreshold = 10; BlockableMonitor monitor = new BlockableMonitor(); try ( OtherThreadExecutor<Void> t2 = new OtherThreadExecutor<>( "T2", null ); BadCollector badCollector = new BadCollector( NULL_OUTPUT_STREAM, UNLIMITED_TOLERANCE, COLLECT_ALL, backPressureThreshold, false, monitor ) ) { for ( int i = 0; i < backPressureThreshold; i++ ) { badCollector.collectDuplicateNode( i, i, "group" ); } Future<Object> enqueue = t2.executeDontWait( command( () -> badCollector.collectDuplicateNode( 999, 999, "group" ) ) ); t2.waitUntilWaiting( waitDetails -> waitDetails.isAt( BadCollector.class, "collect" ) ); monitor.unblock(); enqueue.get(); } }