private SourceModuleContext<?> handleSourceModule(final LocalPhysicalVertex pv, final PullModuleThreadUnit unit,
                                                      final SourceModule<?> instance) {
        createQueues(unit, pv);
        final SourceModuleContext<?> context = (SourceModuleContext<?>) unit.getThreadContext();

        final Vertex v = pv.getVertex();

        for (int i = 0; i < v.getOutputQueueCount(); i++) {
            unitController.allowUsingNewOutputQueue(context, v.getOutputQueueHandle(i).getQueueName(), i,
                                                    pv.getOutputQueueThreadLocalId(i));

            unitController.addThreadOutputRoute(unit, (OutputContext<?>) unit.getThreadContext(), i,
                                                pv.getOutputQueueThreadLocalId(i));
        }

        return context;
    }