class LonelyProcessingStep extends AbstractStep<Void> {

    @Override
    public long receive( long ticket, Void nothing )
    {
        new Thread( () -> {
            assertHealthy();
            try
            {
                try
                {
                    lastProcessingTimestamp = nanoTime();
                    process();
                    endOfUpstream();
                }
                catch ( Throwable e )
                {
                    // we need to update panic state before ending upstream and notifying executor that we completed
                    issuePanic( e );
                }
            }
            catch ( Throwable e )
            {
                // to avoid cases when we hide original panic problem
                // check first if we already in panic state and if so - rethrow original panic cause
                if ( !isPanic() )
                {
                    issuePanic( e );
                }
                else
                {
                    throw e;
                }
            }
        } ).start();
        return 0;
    }

    public  LonelyProcessingStep( StageControl control, String name, Configuration config,
            StatsProvider... additionalStatsProviders );

}