fix: bulk ingester might skip listener requests

[fabriziofortino]

When BulkIngester uses the internal scheduler it might fail notifying the listener because the scheduler gets terminated before the submit is called. This happens because the close condition is signalled before submit is invoked. This PR fixes a regression introduced in #830

[cla-checker-service]

💚 CLA has been signed

[l-trotta]

Hello @fabriziofortino, thank you for the contribution and for highlighting this issue. I have tested your solution and while it helps mitigate the problem, when dealing with many requests there's still a possibility that the latest listener tasks are ignored because of the shutdown. I'll do more testing to figure out a way to handle this reliably, I'll let you know so that we can fix this PR and merge it.

[l-trotta]

@fabriziofortino do you mind if I push on this branch?

[fabriziofortino]

@l-trotta no problem, go ahead!

[fabriziofortino]

@l-trotta I have also noticed this: in the close() method when the scheduler is internal shutdownNow() gets called:

       if (scheduler != null && !isExternalScheduler) {
            scheduler.shutdownNow();
        }

This will try to cancel the already submitted tasks. Shouldn't we call shutdown() instead?

[l-trotta]

@fabriziofortino you're correct, however with the new changes it shouldn't be a problem either way, you can take a look :)

[swallez]

LGTM. Thanks for having caught this and the fix!