Propagates slurp failures to user of processing

so that failures doesn't just hang the processing, but not providing
more jobs and doesn't signal that it has failed.

community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/staging/TicketedProcessing.java

@@ -30,6 +30,7 @@
 import org.neo4j.unsafe.impl.batchimport.Parallelizable;
 import org.neo4j.unsafe.impl.batchimport.executor.DynamicTaskExecutor;
 import org.neo4j.unsafe.impl.batchimport.executor.ParkStrategy;
+import org.neo4j.unsafe.impl.batchimport.executor.TaskExecutionPanicException;
 import org.neo4j.unsafe.impl.batchimport.executor.TaskExecutor;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -91,14 +92,22 @@ public boolean test( long ticket )
     };
     private final Runnable healthCheck;
     private volatile boolean done;
+    private volatile Throwable slurpFailure;
 
     public TicketedProcessing( String name, int maxProcessors, BiFunction<FROM,STATE,TO> processor,
             Supplier<STATE> threadLocalStateSupplier )
     {
         this.processor = processor;
         this.executor = new DynamicTaskExecutor<>( 1, maxProcessors, maxProcessors, park, name,
                 threadLocalStateSupplier );
-        this.healthCheck = executor::assertHealthy;
+        this.healthCheck = () ->
+        {
+            if ( slurpFailure != null )
+            {
+                throw new TaskExecutionPanicException( "Failure adding jobs for processing", slurpFailure );
+            }
+            executor.assertHealthy();
+        };
         this.processed = new ArrayBlockingQueue<>( maxProcessors );
     }
 
@@ -156,15 +165,24 @@ public Future<Void> slurp( Iterator<FROM> input, boolean shutdownAfterAllSubmitt
     {
         return future( () ->
         {
-            while ( input.hasNext() )
+            try
             {
-                submit( input.next() );
+                while ( input.hasNext() )
+                {
+                    submit( input.next() );
+                }
+                if ( shutdownAfterAllSubmitted )
+                {
+                    shutdown( TaskExecutor.SF_AWAIT_ALL_COMPLETED );
+                }
+                return null;
             }
-            if ( shutdownAfterAllSubmitted )
+            catch ( Throwable e )
             {
-                shutdown( TaskExecutor.SF_AWAIT_ALL_COMPLETED );
+                slurpFailure = e;
+                shutdown( TaskExecutor.SF_ABORT_QUEUED );
+                throw e;
             }
-            return null;
         } );
     }
 

community/kernel/src/test/java/org/neo4j/unsafe/impl/batchimport/staging/TicketedProcessingTest.java

@@ -22,19 +22,25 @@
 import org.junit.Rule;
 import org.junit.Test;
 
+import java.util.Iterator;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.BiFunction;
 
 import org.neo4j.test.OtherThreadExecutor.WorkerCommand;
 import org.neo4j.unsafe.impl.batchimport.executor.ParkStrategy;
+import org.neo4j.unsafe.impl.batchimport.executor.TaskExecutionPanicException;
 import org.neo4j.test.OtherThreadRule;
 import org.neo4j.unsafe.impl.batchimport.executor.TaskExecutor;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 import static java.lang.Integer.parseInt;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
@@ -130,6 +136,38 @@ public Void doWork( Void state ) throws Exception
         processing.shutdown( TaskExecutor.SF_AWAIT_ALL_COMPLETED );
     }
 
+    @Test( timeout = 10_000 )
+    public void shouldShutdownWithAbortOnSlurpFailure() throws Exception
+    {
+        // GIVEN
+        TicketedProcessing<StringJob,Void,Integer> processing = new TicketedProcessing<>( "Slurper", 2,
+                (job,state) ->
+                {
+                    return parseInt( job.string );
+                }, () -> null );
+        @SuppressWarnings( "unchecked" )
+        Iterator<StringJob> jobs = mock( Iterator.class );
+        when( jobs.hasNext() ).thenReturn( true );
+        RuntimeException runtimeException = new RuntimeException( "Slurp failure" );
+        when( jobs.next() ).thenReturn( new StringJob( "1" ) ).thenThrow( runtimeException );
+        processing.slurp( jobs, true );
+
+        // WHEN
+        assertEquals( 1, processing.next().intValue() );
+
+        // THEN
+        try
+        {
+            processing.next();
+            fail( "Should have failed with the provided exception" );
+        }
+        catch ( TaskExecutionPanicException e )
+        {
+            // Good
+            assertEquals( runtimeException.getMessage(), e.getCause().getMessage() );
+        }
+    }
+
     private static class StringJob
     {
         final String string;