Input filtering in importer still shows progress

Previously filtering of input data could potentially have large sections
of input be filtered and so now progress displayed. Since both progress
and processor assignments is based on number of processed batches the
behaviour of the iterator batch now changed so that it sends empty no-op
batches for the skipped entries.

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

@@ -33,38 +33,68 @@ public abstract class IteratorBatcherStep<T> extends IoProducerStep
 {
     private final Iterator<T> data;
     private final Class<T> itemClass;
-    protected long cursor;
     private final Predicate<T> filter;
 
+    protected long cursor;
+    private T[] batch;
+    private int batchCursor;
+    private int skipped;
+
     public IteratorBatcherStep( StageControl control, Configuration config, Iterator<T> data, Class<T> itemClass,
             Predicate<T> filter )
     {
         super( control, config );
         this.data = data;
         this.itemClass = itemClass;
         this.filter = filter;
+        newBatch();
+    }
+
+    @SuppressWarnings( "unchecked" )
+    private void newBatch()
+    {
+        batchCursor = 0;
+        batch = (T[]) Array.newInstance( itemClass, batchSize );
     }
 
     @Override
     protected Object nextBatchOrNull( long ticket, int batchSize )
     {
-        @SuppressWarnings( "unchecked" )
-        T[] batch = (T[]) Array.newInstance( itemClass, batchSize );
-        int i = 0;
-        for ( ; i < batchSize && data.hasNext(); )
+        while ( data.hasNext() )
         {
             T candidate = data.next();
             if ( filter.test( candidate ) )
             {
-                batch[i++] = candidate;
+                batch[batchCursor++] = candidate;
                 cursor++;
+                if ( batchCursor == batchSize )
+                {
+                    T[] result = batch;
+                    newBatch();
+                    return result;
+                }
+            }
+            else
+            {
+                if ( ++skipped == batchSize )
+                {
+                    skipped = 0;
+                    return Array.newInstance( itemClass, 0 );
+                }
             }
         }
 
-        if ( i == 0 )
+        if ( batchCursor == 0 )
         {
             return null; // marks the end
         }
-        return i == batchSize ? batch : Arrays.copyOf( batch, i );
+        try
+        {
+            return batchCursor == batchSize ? batch : Arrays.copyOf( batch, batchCursor );
+        }
+        finally
+        {
+            batchCursor = 0;
+        }
     }
 }