Simplify the BufferedAsyncIterator

[ankon]

The goal of that iterator is to take a number of promises in unknown states, and generate
their results in the order of these promises resolving/rejecting.

This commit changes the implementation in such a way that we do not need an event emitter and
a result queue and a (fairly complex) logic of synchronizing these. Instead the iterator
builds two parallel queues, one containing unresolved promises and one containing the [resolve, reject]
functions for these promises. The incoming promises then are configured to pop the next [resolve, reject]
pair, and call that with the respective result. The iterator at the same time picks the next unresolved
promise from its own queue, and awaits the result.

On the runner side this requires to await the iterator.next(), which actually simplifies more code
by removing the need for a separate queue of "enqueued tasks".

Note that this code also removes the configurable "handleError" logic without replacement: No caller
ever provided that, so the default of "throw the error further" is good enough.

[ankon]

This works in my local tests fine, but given that I have issues getting the KafkaJS tests to run reliably I hope the Azure checks will show that the unit/integration tests also pass.

NOTE: This is currently on top of master, i.e. the bug fixed with #670 is still here

[ankon]

Note here: I changed batches.map to batches.forEach, because map implies that the result is used, but here the side-effects are what count and the result is ignored.

[ankon]

Rebased onto master to resolve the conflict introduced with #670 .

I think it would be good to pull this in before 1.13 if possible, as it simplifies code introduced in 1.13 -- so less variation in the releases :)

[goriunov]

Was just comparing with my PR #705 :) TBH i i think finally is not needed here as if we fail it will call unlockWithError and even if we later call unlock nothing will happened as promise has been rejected.

[goriunov]

I was abit concerned about awaiting in the while loop as it is not going to wait for this and will just loop and saturate cpu... (this is just assumption have not validate that...)

[ankon]

The await-ing here is needed because the generator is now doing funny async stuff as well. It shouldn't be a busy loop though, but rather it awaits here for the next promise in the bufferedasynciterator to resolve.

[goriunov]

Probably would be good idea to join some of my changes from #705 with yours ! 😄

[ankon]

Probably would be good idea to join some of my changes from #705 with yours ! 😄

Indeed, that would be my plan as well. Given that #705 fixes an actual problem, I'd think the best approach here could be to wait for #705 to land, and then to rebase this one. WDYT?

[ankon]

Indeed, that would be my plan as well. Given that #705 fixes an actual problem, I'd think the best approach here could be to wait for #705 to land, and then to rebase this one. WDYT?

Small update: #705 is replaced by #714 of course, and it seems that merging these together won't be much of an issue. We're now (temporarily) using our own fork of KafkaJS with both changes applied, as the changes in the beta are too tempting to not use and they do fix at least one issue for us. Our fork is at https://github.com/Collaborne/kafkajs/tree/collaborne, and can be used for reference (but obviously shouldn't be used in production by anyone unless you know what you're doing -- it's a prerelease, after all!)

[goriunov]

@ankon just went through this change again and noticed that after your change requests are not processed asynchronously any more as it will wait for every single request to be processed in order, before they were processed asynchronously in Promise.all().

[ankon]

Hmm.

I'll have a look, that's certainly not the idea here. Good hint to maybe write a test showing that it indeed processes things asynchronously. :)

[ankon]

While looking further here I find the existing code harder and harder to reason about.

1. There is probably a way to reject a promise twice, because the event emitter's error listener isn't removed (while the data listener is). This is likely benign, as per http://www.ecma-international.org/ecma-262/6.0/#sec-promise-reject-functions, but confusing when reading.
2. I'm thinking that there is also an opportunity for a race where an error might get eaten. This probably requires very careful timing and some asynchronicity in processing batches to trigger, but assuming that there are results in the queue and the next promise is quick enough to error it seems that there won't be an error handler installed for the error -- except for the ones left over from the previous problem, of course. :)
3. While it is true that it looks "more concurrent" (less "await"-y), it's difficult to see exactly where stuff is getting blocked on which conditions. At the end we have the async iterator doing things, the consumer group/runner doing things, and of course the concurrency-limiting also doing its thing.

That might all be me missing something while wrapping my head around this again of course.

On the good news side: There is already tests for the concurrency behavior of the iterator, and I definitely missed to adapt these -- so I'll see whether I can extend these further as well.

[ankon]

I went over the changes now and rebased them onto master.

I also decided to remove the change from a generator to an async generator for now, because as @goriunov pointed out there is definitely a change in how/where the asynchronicity happens. This makes this change simpler, and more obviously a refactoring: The interface of BufferedAsyncIterator stays the same, the tests stay the same (and still pass :D), but the whole thing lost half of its lines and hopefully also got a lot more easy to process as a human.

This PR now has two smaller JSDoc improvements to sprinkle some type information which helps when reading this code as well, tell me if I better create separate PRs for these.

[t-d-d]

+1. I don't understand the original code but do understand this.

[Nevon]

I wasn't quite sure about the implications of this, since the code is quite complex and it's easy to make a change that accidentally changes the asynchronous behavior, so I wrote a test that uses trace_events to create a timeline for how the promises behave. The test setup itself was simply:

const BufferedAsyncIterator = require(`./${process.argv[2]}.js`)
const tracing = trace_events.createTracing({ categories: ['node.async_hooks'] })
tracing.enable();

const promises = [sleep(300).then(() => 1), sleep(100).then(() => 2), sleep(500).then(() => 3)]
const iterator = BufferedAsyncIterator(promises)

while (true) {
    const result = iterator.next()
    if (result.done) {
        break
    }

    await result.value
}
tracing.disable()

And then I ran it with both implementations and compared the results in Chrome's trace view. Left is the original implementation and right is the simplified one.

As you can see, the behavior is identical between the different implementations.

The traces are available here:
original-50845-1.log
simplified-50858-1.log

[ankon]

(Rebased onto current master)

[Nevon]

Did some more checks to see that error recovery was working the same way, and everything seems good. I also find this implementation a lot easier to understand, so I'm gonna go ahead and merge now that I'm confident that it's not changing the behavior. Thanks for the refactor!