public void error( Throwable throwable ) {
    publishProcessor.onError( throwable );
  }
@Test
  public void testError() {
    // verifies that calling .error() results in an exception being thrown
    // by the .rows() blocking iterable.

    final String exceptionMessage = "Exception raised during acceptRows loop";
    streamSource = new BlockingQueueStreamSource<String>( streamStep ) {
      @Override public void open() {
        execSvc.submit( () -> {
          for ( int i = 0; i < 10; i++ ) {
            acceptRows( singletonList( "new row " + i ) );
            try {
              Thread.sleep( 5 );
            } catch ( InterruptedException e ) {
              fail();
            }
            if ( i == 5 ) {

              error( new RuntimeException( exceptionMessage ) );
              break;
            }
          }
        } );
      }
    };
    streamSource.open();
    Iterator<String> iterator = streamSource.flowable().blockingIterable().iterator();

    Future<List<String>> iterLoop = execSvc.submit( () -> {
      List<String> strings = new ArrayList<>();
      do {
        strings.add( iterator.next() );
      } while ( strings.size() < 9 );
      return strings;
    } );
    try {
      iterLoop.get( 50, MILLISECONDS );
      fail( "expected exception" );
    } catch ( InterruptedException | ExecutionException | TimeoutException e ) {
      // occasionally fails wingman with npe for mysterious reason,
      // so guarding with null check
      if ( e != null && e.getCause() != null ) {
        assertThat( e.getCause().getMessage(), equalTo( exceptionMessage ) );
      }
    }
  }